274 lines
12 KiB
Python
274 lines
12 KiB
Python
"""
|
||
Scholengroep: leerlingenaantallen van alle scholen in PIVOT.
|
||
|
||
Elke school wordt geteld met haar EIGEN Informat-koppeling (instellingsnummer,
|
||
client-ID en secret uit het scholenregister). Het resultaat wordt bewaard in de
|
||
centrale database, zodat elke school het overzicht van de scholengroep ziet.
|
||
|
||
- Huidig aantal: peildatum vandaag + 1 maand (kolommen in Scholen).
|
||
- Telling op peildatum: een gekozen datum (bv. 1 februari), bewaard in Leerlingenaantal_tellingen,
|
||
zodat je ze later opnieuw kan bekijken en exporteren.
|
||
"""
|
||
import csv
|
||
import io
|
||
import json
|
||
from datetime import date, datetime
|
||
|
||
from flask import Blueprint, render_template, request, flash, redirect, url_for, session, Response
|
||
from blueprints.auth import login_required
|
||
from database import central_query
|
||
from rechten import recht_vereist
|
||
from services.sync_service import SyncService
|
||
from tenancy import alle_scholen, school_op_code, in_school, herlaad, ONDERWIJSNIVEAUS
|
||
|
||
leerlingenaantal_bp = Blueprint('leerlingenaantal', __name__)
|
||
|
||
INFORMAT_VELDEN = ('INFORMAT_INST_NUMBER', 'INFORMAT_ID', 'INFORMAT_SECRET')
|
||
|
||
# Teksten en routes voor het gedeelde sjabloon leerlingenaantal.html (ook gebruikt door de personeelsaantallen)
|
||
WEERGAVE = dict(ep='leerlingenaantal', pagina='leerlingenaantal_page', titel='Leerlingenaantallen',
|
||
eenheid='Leerlingen', kort='lln.')
|
||
|
||
|
||
def _datum(waarde):
|
||
try:
|
||
return datetime.strptime((waarde or '').strip(), '%Y-%m-%d').date()
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _fmt(dt, leeg=''):
|
||
if not dt:
|
||
return leeg
|
||
return dt.strftime('%d-%m-%Y %H:%M') if hasattr(dt, 'strftime') else str(dt)
|
||
|
||
|
||
def _detail(tekst):
|
||
try:
|
||
return json.loads(tekst) if tekst else []
|
||
except ValueError:
|
||
return []
|
||
|
||
|
||
NIVEAUS = list(ONDERWIJSNIVEAUS) + [('', 'Onderwijsniveau niet ingesteld')]
|
||
NIVEAU_LABEL = dict(NIVEAUS)
|
||
|
||
|
||
def _niveau(school):
|
||
n = school.waarden.get('ONDERWIJSNIVEAU') or ''
|
||
return n if n in NIVEAU_LABEL else ''
|
||
|
||
|
||
def _som(rijen, veld):
|
||
return sum(r[veld] for r in rijen if isinstance(r[veld], int))
|
||
|
||
|
||
def _groepen(rows, veld):
|
||
"""Rijen per onderwijsniveau (basis, secundair, niet ingesteld) met subtotaal; lege groepen vallen weg."""
|
||
uit = []
|
||
for code, label in NIVEAUS:
|
||
rijen = [r for r in rows if r['Niveau'] == code]
|
||
if rijen:
|
||
uit.append({'code': code, 'label': label, 'rijen': rijen, 'subtotaal': _som(rijen, veld)})
|
||
return uit
|
||
|
||
|
||
def _huidig(query=''):
|
||
rows = []
|
||
for s in alle_scholen():
|
||
if query and query.lower() not in s.naam.lower():
|
||
continue
|
||
rows.append({
|
||
'Code': s.code,
|
||
'Naam_school': s.naam,
|
||
'Niveau': _niveau(s),
|
||
'Leerlingenaantal': s.leerlingenaantal if s.leerlingenaantal is not None else '–',
|
||
'Huidig': s.leerlingenaantal if s.leerlingenaantal is not None else '–',
|
||
'Laatste_synchronisatie': s.leerlingenaantal_sync,
|
||
'Informat': s.heeft(*INFORMAT_VELDEN),
|
||
'Campussen': s.leerlingenaantal_detail or [], # [{'naam', 'nummer', 'aantal'}] bij meerdere campussen
|
||
})
|
||
return rows
|
||
|
||
|
||
def _telling(peildatum, query=''):
|
||
"""Alle actieve scholen met hun aantal op de peildatum (None = niet geteld)."""
|
||
per = {r['School_ID']: r for r in central_query(
|
||
"SELECT School_ID, Aantal, Detail, Schooljaar, Opgevraagd_op, Door FROM Leerlingenaantal_tellingen "
|
||
"WHERE Peildatum = %s", (peildatum,))}
|
||
rows = []
|
||
for s in alle_scholen():
|
||
if query and query.lower() not in s.naam.lower():
|
||
continue
|
||
t = per.get(s.id) or {}
|
||
rows.append({'Code': s.code, 'Naam_school': s.naam, 'Niveau': _niveau(s), 'Aantal': t.get('Aantal'),
|
||
'Schooljaar': t.get('Schooljaar'), 'Opgevraagd_op': t.get('Opgevraagd_op'),
|
||
'Door': t.get('Door'), 'Informat': s.heeft(*INFORMAT_VELDEN),
|
||
'Campussen': _detail(t.get('Detail'))})
|
||
return rows
|
||
|
||
|
||
def _peildata():
|
||
return central_query("SELECT Peildatum, COUNT(*) AS Scholen, SUM(Aantal) AS Totaal, MAX(Opgevraagd_op) AS Laatst "
|
||
"FROM Leerlingenaantal_tellingen GROUP BY Peildatum ORDER BY Peildatum DESC")
|
||
|
||
|
||
@leerlingenaantal_bp.route("/leerlingenaantal", methods=["GET", "POST"])
|
||
@login_required
|
||
@recht_vereist('leerlingenaantal')
|
||
def leerlingenaantal_page():
|
||
query = (request.values.get("query") or "").strip()
|
||
herlaad(force=True)
|
||
results = _huidig(query)
|
||
totaal = _som(results, 'Leerlingenaantal')
|
||
|
||
peildatum = _datum(request.args.get('peildatum'))
|
||
telling = _telling(peildatum, query) if peildatum else None
|
||
telling_totaal = _som(telling, 'Aantal') if telling else 0
|
||
return render_template("leerlingenaantal.html", results=results, query=query, totaal=totaal,
|
||
groepen=_groepen(results, 'Leerlingenaantal'),
|
||
telling_groepen=_groepen(telling, 'Aantal') if telling else [],
|
||
peildatum=peildatum, telling=telling, telling_totaal=telling_totaal,
|
||
peildata=_peildata(), vandaag=date.today(), **WEERGAVE)
|
||
|
||
|
||
@leerlingenaantal_bp.route("/sync-aanvragen", methods=["POST"])
|
||
@login_required
|
||
@recht_vereist('leerlingenaantal')
|
||
def sync_aanvragen():
|
||
codes = [s.code for s in alle_scholen()] if request.form.get("alle") else [request.form.get("school_code")]
|
||
for code in codes:
|
||
school = school_op_code(code)
|
||
if not school:
|
||
continue
|
||
if not school.heeft(*INFORMAT_VELDEN):
|
||
flash(f"{school.naam}: Informat is nog niet ingesteld.", "error")
|
||
continue
|
||
with in_school(school):
|
||
success, message = SyncService().sync_student_count()
|
||
flash(f"{school.naam}: {message}", "success" if success else "error")
|
||
return redirect(url_for('leerlingenaantal.leerlingenaantal_page'))
|
||
|
||
|
||
@leerlingenaantal_bp.route("/leerlingenaantal/peildatum", methods=["POST"])
|
||
@login_required
|
||
@recht_vereist('leerlingenaantal')
|
||
def tel_peildatum():
|
||
"""Telt de leerlingen op de gekozen datum, voor alle scholen of één school."""
|
||
peildatum = _datum(request.form.get('peildatum'))
|
||
if not peildatum:
|
||
flash("Kies een geldige datum.", "error")
|
||
return redirect(url_for('leerlingenaantal.leerlingenaantal_page'))
|
||
code = request.form.get('school_code')
|
||
scholen = [school_op_code(code)] if code else alle_scholen()
|
||
door = session.get('user')
|
||
gelukt = 0
|
||
for school in scholen:
|
||
if not school:
|
||
continue
|
||
if not school.heeft(*INFORMAT_VELDEN):
|
||
if code:
|
||
flash(f"{school.naam}: Informat is nog niet ingesteld.", "error")
|
||
continue
|
||
with in_school(school):
|
||
ok, res = SyncService().tel_op_peildatum(peildatum, door)
|
||
if ok:
|
||
gelukt += 1
|
||
else:
|
||
flash(f"{school.naam}: {res}", "error")
|
||
if gelukt:
|
||
flash(f"Leerlingen geteld op {peildatum:%d-%m-%Y} voor {gelukt} school/scholen.", "success")
|
||
return redirect(url_for('leerlingenaantal.leerlingenaantal_page', peildatum=peildatum.isoformat()))
|
||
|
||
|
||
@leerlingenaantal_bp.route("/leerlingenaantal/peildatum/verwijderen", methods=["POST"])
|
||
@login_required
|
||
@recht_vereist('leerlingenaantal')
|
||
def verwijder_peildatum():
|
||
peildatum = _datum(request.form.get('peildatum'))
|
||
if peildatum:
|
||
central_query("DELETE FROM Leerlingenaantal_tellingen WHERE Peildatum = %s", (peildatum,), commit=True)
|
||
flash(f"Telling van {peildatum:%d-%m-%Y} verwijderd.", "success")
|
||
return redirect(url_for('leerlingenaantal.leerlingenaantal_page'))
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# CSV
|
||
# ------------------------------------------------------------------
|
||
def _csv(naam, kop, rijen):
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf, delimiter=';') # ; zodat Excel (Belgische instellingen) de kolommen meteen splitst
|
||
w.writerow(kop)
|
||
w.writerows(rijen)
|
||
return Response('' + buf.getvalue(), mimetype='text/csv',
|
||
headers={'Content-Disposition': f'attachment; filename={naam}'})
|
||
|
||
|
||
@leerlingenaantal_bp.route("/leerlingenaantal/export.csv", methods=["GET"])
|
||
@login_required
|
||
@recht_vereist('leerlingenaantal')
|
||
def export_csv():
|
||
"""?peildatum=YYYY-MM-DD → die telling; ?alle=1 → alle peildata naast elkaar; anders het huidige aantal."""
|
||
query = (request.args.get('query') or '').strip()
|
||
|
||
if request.args.get('alle'):
|
||
data = central_query("SELECT School_ID, Peildatum, Aantal, Detail FROM Leerlingenaantal_tellingen ORDER BY Peildatum")
|
||
datums = sorted({r['Peildatum'] for r in data})
|
||
per = {(r['School_ID'], r['Peildatum']): r['Aantal'] for r in data}
|
||
# campus-aantallen: {(school_id, campusnaam, datum): aantal}
|
||
campus, campusnamen = {}, {}
|
||
for r in data:
|
||
for d in _detail(r['Detail']):
|
||
campus[(r['School_ID'], d['naam'], r['Peildatum'])] = d['aantal']
|
||
campusnamen.setdefault(r['School_ID'], [])
|
||
if d['naam'] not in campusnamen[r['School_ID']]:
|
||
campusnamen[r['School_ID']].append(d['naam'])
|
||
scholen = [s for s in alle_scholen() if not query or query.lower() in s.naam.lower()]
|
||
rijen = []
|
||
for code, label in NIVEAUS:
|
||
groep = [s for s in scholen if _niveau(s) == code]
|
||
if not groep:
|
||
continue
|
||
for s in groep:
|
||
rijen.append([label, s.naam, ''] + [per.get((s.id, d), '') for d in datums])
|
||
for naam in campusnamen.get(s.id, []):
|
||
rijen.append([label, s.naam, naam] + [campus.get((s.id, naam, d), '') for d in datums])
|
||
if code:
|
||
rijen.append([f"Subtotaal {label}", '', ''] + [sum(per.get((s.id, d)) or 0 for s in groep) for d in datums])
|
||
rijen.append(['Totaal', '', ''] + [sum(per.get((s.id, d)) or 0 for s in scholen) for d in datums])
|
||
return _csv(f"leerlingenaantallen_alle_peildata_{date.today():%Y%m%d}.csv",
|
||
['Onderwijsniveau', 'School', 'Campus'] + [f"{d:%d-%m-%Y}" for d in datums], rijen)
|
||
|
||
peildatum = _datum(request.args.get('peildatum'))
|
||
if peildatum:
|
||
rows = _telling(peildatum, query)
|
||
rijen = []
|
||
for g in _groepen(rows, 'Aantal'):
|
||
for r in g['rijen']:
|
||
rijen.append([g['label'], r['Naam_school'], '', '' if r['Aantal'] is None else r['Aantal'],
|
||
r['Schooljaar'] or '', _fmt(r['Opgevraagd_op'], 'niet geteld'), r['Door'] or ''])
|
||
for c in r['Campussen']:
|
||
rijen.append([g['label'], r['Naam_school'], c['naam'], c['aantal'], r['Schooljaar'] or '', '', ''])
|
||
if g['code']:
|
||
rijen.append([f"Subtotaal {g['label']}", '', '', g['subtotaal'], '', '', ''])
|
||
rijen.append(['Totaal', '', '', _som(rows, 'Aantal'), '', '', ''])
|
||
return _csv(f"leerlingenaantallen_{peildatum:%Y%m%d}.csv",
|
||
['Onderwijsniveau', 'School', 'Campus', f"Leerlingen op {peildatum:%d-%m-%Y}", 'Schooljaar',
|
||
'Opgevraagd op', 'Door'], rijen)
|
||
|
||
rows = _huidig(query)
|
||
rijen = []
|
||
for g in _groepen(rows, 'Leerlingenaantal'):
|
||
for r in g['rijen']:
|
||
rijen.append([g['label'], r['Naam_school'], '',
|
||
r['Leerlingenaantal'] if isinstance(r['Leerlingenaantal'], int) else '',
|
||
_fmt(r['Laatste_synchronisatie'], 'nog nooit')])
|
||
for c in r['Campussen']:
|
||
rijen.append([g['label'], r['Naam_school'], c['naam'], c['aantal'], ''])
|
||
if g['code']:
|
||
rijen.append([f"Subtotaal {g['label']}", '', '', g['subtotaal'], ''])
|
||
rijen.append(['Totaal', '', '', _som(rows, 'Leerlingenaantal'), ''])
|
||
return _csv(f"leerlingenaantallen_{date.today():%Y%m%d}.csv",
|
||
['Onderwijsniveau', 'School', 'Campus', 'Leerlingen', 'Laatste synchronisatie'], rijen)
|
||
|