v0.0.5: gedeelde mailboxen (info@, facturen@, nieuwsbrief@, persoonlijk) met eigen IMAP/SMTP, afzender, rol en toegang per gebruiker; inbox per mailbox en versturen vanuit een gekozen mailbox; routines (jobs.py: types met @jobs.register, interval, instellingen, logboek, slot tegen dubbel draaien; app.py scheduler/tick) voor e-mail ophalen, nieuwsbrieven versturen en opschonen; meldingen (bel, bureaubladmelding); nieuwsbrieven met segmenten, testmail, inplannen en versturen per batch over één verbinding, afmeldlink met List-Unsubscribe en bounceverwerking; oude mailinstellingen worden de eerste mailbox

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Machiel
2026-10-16 18:25:44 +02:00
co-authored by Claude Opus 5.5
parent 8f6e1ab443
commit cad838f1f4
24 changed files with 1710 additions and 185 deletions
+88
View File
@@ -0,0 +1,88 @@
"""Routines: terugkerende taken van de CRM (e-mail ophalen, nieuwsbrieven versturen, …).
Een routinetype is een functie met @register; een routine (tabel jobs) is een ingestelde instantie van een type
met een interval en eigen instellingen (config, JSON). Er kunnen meerdere routines van hetzelfde type zijn, bijv.
twee keer e-mail ophalen met andere mailboxen en een ander interval.
Uitvoeren: `python app.py scheduler` (blijft draaien, kijkt elke 30 s; de Docker-container crm-routines) of
`python app.py tick` (één keer, voor cron elke minuut). Een routine draait nooit twee keer tegelijk (running_since
werkt als slot; na STALE_MINUTES geldt een slot als achtergebleven na een crash).
Een nieuw type toevoegen, in app.py:
@jobs.register('rapport', 'Weekrapport', 'Mailt elke week de nieuwe leads.', interval=7 * 24 * 60,
fields=[jobs.Field('to', 'Naar', 'email')])
def job_rapport(job, config):
... # gewone app-context: db(), q(), notify(), mail…
return '12 leads gemeld' # komt in het logboek; een exception = status fout
Een standaardroutine die er altijd moet zijn, krijgt `default=True`: die maakt app.py bij het starten één keer
aan (actief, met het standaardinterval).
"""
from dataclasses import dataclass, field
STALE_MINUTES = 60
KEEP_RUNS = 200 # logregels per routine
UNITS = [(1, 'minuten'), (60, 'uur'), (24 * 60, 'dagen')]
@dataclass
class Field:
"""Instelling van een routinetype. kind: text, number, email, bool, mailboxes (meerkeuze uit de mailboxen)."""
key: str
label: str
kind: str = 'text'
default: object = ''
hint: str = ''
@dataclass
class JobType:
key: str
label: str
description: str
run: object
interval: int = 60 # standaardinterval in minuten
fields: list = field(default_factory=list)
default: bool = False # bij het starten één keer aanmaken
TYPES = {}
def register(key, label, description='', interval=60, fields=(), default=False):
def wrap(fn):
TYPES[key] = JobType(key, label, description, fn, interval, list(fields), default)
return fn
return wrap
def defaults(kind):
return {f.key: f.default for f in TYPES[kind].fields}
def clean_config(kind, form):
"""Instellingen uit een formulier (request.form) volgens de velden van het type."""
out = {}
for f in TYPES[kind].fields:
if f.kind == 'bool':
out[f.key] = bool(form.get(f.key))
elif f.kind == 'number':
try:
out[f.key] = int(form.get(f.key) or f.default or 0)
except ValueError:
out[f.key] = f.default
elif f.kind == 'mailboxes':
out[f.key] = [int(v) for v in form.getlist(f.key) if v.isdigit()]
else:
out[f.key] = (form.get(f.key) or '').strip()
return out
def interval_text(minutes):
for size, unit in reversed(UNITS):
if minutes % size == 0 and minutes >= size:
n = minutes // size
return {'minuten': f'{n} min', 'uur': f'{n} uur', 'dagen': f'{n} dag' + ('en' if n != 1 else '')}[unit]
return f'{minutes} min'