""" LGCC Alerts Module v1.0 — Fase B PVA "Proactieve Agendabesturing" ================================================================== Proactieve signalering op basis van de vastgestelde agenda (bron van waarheid: data/agenda-110AN-2026-2027.json óf de agenda_items-tabel uit Fase A). Signaalschema (vastgesteld Jorick, 17 aug 2026): T-20 / T-14 / T-7 / T-2 Stiltevenster: nooit pushes vóór 08:00 Europe/Amsterdam (afgehandeld in telegram_push.py; dit module berekent alleen). Functionaliteit: - bereken_signalen(vandaag=None): welke signalen gaan vandaag af + welke items liggen binnen 14 dagen; schrijft data/alerts-vandaag.json (idempotent) - route GET /agenda/api/alerts (JSON: actieve alerts + komende items) Integratie in app.py (zelfde pattern als agenda_module / newsletter_module): from alerts_module import register_alert_routes register_alert_routes(app, page, BASE_PATH, db) Lokaal testen (repo): python3 -m pytest tests/test_alerts.py -v Standalone (cron/berekening buiten Flask om): python3 alerts_module.py # herberekent alerts-vandaag.json voor vandaag python3 alerts_module.py 2026-09-01 # voor een specifieke datum """ from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo import json import os import sys MODULE_VERSION = '1.0' TZ_AMSTERDAM = ZoneInfo('Europe/Amsterdam') # Vastgesteld signaalschema: label -> dagen vooraf SIGNAL_SCHEMA = {'T-20': 20, 'T-14': 14, 'T-7': 7, 'T-2': 2} KOMENDE_DAGEN = 14 # horizon "komende items" MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.environ.get('ALERTS_DATA_DIR') or os.path.join(MODULE_DIR, 'data') AGENDA_JSON = os.path.join(DATA_DIR, 'agenda-110AN-2026-2027.json') ALERTS_JSON = os.path.join(DATA_DIR, 'alerts-vandaag.json') # ────────────────────────────────────────────────────────────────────────────── # Bronnen: JSON-file óf agenda_items-tabel (Fase A) — beide ondersteund # ────────────────────────────────────────────────────────────────────────────── def _laad_items_uit_json(pad=None): """Lees agenda-items uit de JSON-bron (met voorberekende signalen-arrays).""" with open(pad or AGENDA_JSON, encoding='utf-8') as f: data = json.load(f) items = [] for it in data.get('items', []): items.append({ 'datum': it['datum'], 'datum_eind': it.get('datum_eind') or None, 'titel': it['titel'], 'type': it.get('type', ''), 'signalen': list(it.get('signalen') or []), 'status': it.get('status', 'open'), }) return items def _laad_items_uit_db(db_module): """Lees agenda-items uit de agenda_items-tabel (Fase A). Signaaldaten worden hier berekend uit de startdatum (T-20/T-14/T-7/T-2). Retourneert None als de tabel niet bestaat, zodat de JSON-bron als fallback kan dienen.""" conn = db_module.get_db() try: bestaat = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='agenda_items'" ).fetchone() if not bestaat: return None rows = conn.execute( "SELECT datum, datum_eind, titel, type, status FROM agenda_items " "ORDER BY datum ASC, titel ASC").fetchall() items = [] for r in rows: items.append({ 'datum': r['datum'], 'datum_eind': (r['datum_eind'] or None) or None, 'titel': r['titel'], 'type': r['type'] or '', 'signalen': [], # wordt berekend uit de startdatum 'status': r['status'] or 'open', }) return items finally: conn.close() def _signaaldatum_voor_label(datum_iso, label): """Signaaldatum = startdatum - x dagen (T-20 → datum-20).""" return (date.fromisoformat(datum_iso) - timedelta(days=SIGNAL_SCHEMA[label])).isoformat() def _signaalmap_voor_item(item): """{signaaldatum: label} voor één item. Signalen gaan altijd uit van de STARTdatum — ook bij meerdaagse items (datum_eind), conform PVA. Als de JSON-bron een vastgestelde signalen-array bevat, wordt die gebruikt (gefilterd op schema-geldige datums); anders wordt het schema berekend.""" map_ = {lbl: _signaaldatum_voor_label(item['datum'], lbl) for lbl in SIGNAL_SCHEMA} if item.get('signalen'): vastgesteld = {lbl: d for lbl, d in map_.items() if d in set(item['signalen'])} if vastgesteld: return vastgesteld return map_ def _is_afgerond(item): return str(item.get('status') or 'open').lower() == 'afgerond' # ────────────────────────────────────────────────────────────────────────────── # KERN — bereken_signalen # ────────────────────────────────────────────────────────────────────────────── def bereken_signalen(vandaag=None, agenda_json=None, db=None, output_pad=None): """ Bepaalt voor `vandaag` (date of 'YYYY-MM-DD'; default: vandaag Amsterdam): - alerts: items waarvan vandaag een signaal afgaat (T-20/T-14/T-7/T-2) - komende: items met startdatum binnen 14 dagen, plus lopende meerdaagse items (datum <= vandaag <= datum_eind) Bron: agenda_items-tabel via `db` indien meegegeven én aanwezig, anders de JSON-file (agenda-110AN-2026-2027.json). Schrijft het resultaat naar `output_pad` (default data/alerts-vandaag.json) — idempotent: herberekenen op dezelfde dag overschrijft dat bestand. Retourneert het resultaat-dict (zelfde inhoud als het JSON-bestand). """ if vandaag is None: vandaag = datetime.now(TZ_AMSTERDAM).date() elif isinstance(vandaag, str): vandaag = date.fromisoformat(vandaag) vandaag_iso = vandaag.isoformat() horizon = (vandaag + timedelta(days=KOMENDE_DAGEN)).isoformat() # ── bronkeuze: DB eerst, JSON als fallback ──────────────────────────── bron, items = 'json', None if db is not None: try: items = _laad_items_uit_db(db) if items is not None: bron = 'database' except Exception: items = None if items is None: items = _laad_items_uit_json(agenda_json) # ── signalen die vandaag afgaan ─────────────────────────────────────── alerts = [] for it in items: if _is_afgerond(it): continue # afgeronde items signaleren niet meer for lbl, signaaldatum in _signaalmap_voor_item(it).items(): if signaaldatum == vandaag_iso: alerts.append({ 'datum': it['datum'], 'datum_eind': it['datum_eind'], 'titel': it['titel'], 'type': it['type'], 'signaaltype': lbl, 'dagen_tot': SIGNAL_SCHEMA[lbl], }) alerts.sort(key=lambda a: (a['datum'], SIGNAL_SCHEMA[a['signaaltype']])) # ── komende items binnen 14 dagen (+ lopende meerdaagse) ────────────── komende = [] for it in items: if _is_afgerond(it): continue d = date.fromisoformat(it['datum']) lopend = bool(it['datum_eind']) and date.fromisoformat(it['datum_eind']) >= vandaag and d <= vandaag if (vandaag_iso <= it['datum'] <= horizon) or lopend: komende.append({ 'datum': it['datum'], 'datum_eind': it['datum_eind'], 'titel': it['titel'], 'type': it['type'], 'dagen_tot': (d - vandaag).days, 'status': it.get('status', 'open'), }) komende.sort(key=lambda k: (k['datum'], k['titel'])) resultaat = { 'datum': vandaag_iso, 'gegenereerd': datetime.now(TZ_AMSTERDAM).isoformat(timespec='seconds'), 'bron': bron, 'signaalschema': list(SIGNAL_SCHEMA.keys()), 'aantal_alerts': len(alerts), 'alerts': alerts, 'aantal_komende': len(komende), 'komende': komende, } pad = output_pad or ALERTS_JSON os.makedirs(os.path.dirname(pad), exist_ok=True) # idempotent: atomair overschrijven — herberekening van dezelfde dag # vervangt het bestand (geen duplicaten, geen historie-opstapeling) tmp = pad + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(resultaat, f, ensure_ascii=False, indent=1) os.replace(tmp, pad) return resultaat # ────────────────────────────────────────────────────────────────────────────── # FLASK-ROUTES — zelfde registreerpatroon als agenda_module (Fase A) # ────────────────────────────────────────────────────────────────────────────── def register_alert_routes(app, page, BASE_PATH, db): """ Registreert de alerts-routes op de Flask-app. `db` is de database-module (met get_db()); mag None zijn — dan valt de module terug op de JSON-bron. """ from flask import request, jsonify @app.route('/agenda/api/alerts', methods=['GET']) def agenda_api_alerts(): """JSON: vandaag actieve signalen + komende items (binnen 14 dagen). Optioneel ?datum=YYYY-MM-DD (admin/debug/cron-herberekening).""" datum_param = request.args.get('datum') if datum_param: try: date.fromisoformat(datum_param) except ValueError: return jsonify({'error': 'Ongeldige datum, verwacht YYYY-MM-DD'}), 400 try: resultaat = bereken_signalen(vandaag=datum_param, db=db) except FileNotFoundError: return jsonify({'error': 'Agenda-bron niet gevonden (JSON of agenda_items-tabel)'}), 500 except Exception as e: return jsonify({'error': f'Berekening mislukt: {str(e)[:150]}'}), 500 return jsonify({ 'datum': resultaat['datum'], 'aantal_alerts': resultaat['aantal_alerts'], 'alerts': resultaat['alerts'], 'aantal_komende': resultaat['aantal_komende'], 'komende': resultaat['komende'], 'bron': resultaat['bron'], }) # ────────────────────────────────────────────────────────────────────────────── # CLI — standalone berekening (cron: elke nacht vóór 08:00 Amsterdam) # ────────────────────────────────────────────────────────────────────────────── if __name__ == '__main__': datum_arg = sys.argv[1] if len(sys.argv) > 1 else None res = bereken_signalen(vandaag=datum_arg) print(f"✅ {res['aantal_alerts']} signaal(en) op {res['datum']} (bron: {res['bron']}) " f"→ {ALERTS_JSON}") for a in res['alerts']: print(f" [{a['signaaltype']}] {a['titel']} ({a['datum']})")