233 lines
11 KiB
Python
233 lines
11 KiB
Python
"""
|
||
Personeel › Personeelsaantallen: hetzelfde overzicht als de leerlingenaantallen, maar voor het personeel.
|
||
|
||
Elke school wordt geteld met haar eigen Informat-koppeling (Personeel-API, zie services/gopro_service.py):
|
||
een personeelslid telt mee als het op de peildatum een lopende opdracht heeft (één keer per persoon).
|
||
- Huidig aantal: vandaag, bewaard in Personeelsaantal_huidig.
|
||
- Telling op peildatum: bewaard in Personeelsaantal_tellingen, later opnieuw te bekijken en te exporteren.
|
||
Het sjabloon is gedeeld met de leerlingenaantallen (leerlingenaantal.html).
|
||
"""
|
||
import csv
|
||
import io
|
||
import json
|
||
from datetime import date
|
||
|
||
from flask import Blueprint, render_template, request, flash, redirect, url_for, session, Response
|
||
from blueprints.auth import login_required
|
||
from blueprints.leerlingenaantal_bp import (INFORMAT_VELDEN, NIVEAUS, _datum, _detail, _fmt, _groepen, _niveau,
|
||
_som)
|
||
from database import central_query
|
||
from rechten import recht_vereist
|
||
from services.gopro_service import GoProService
|
||
from tenancy import alle_scholen, school_op_code, in_school, _automatisch_schooljaar
|
||
|
||
personeelsaantal_bp = Blueprint('personeelsaantal', __name__)
|
||
|
||
WEERGAVE = dict(ep='personeelsaantal', pagina='personeelsaantal_page', titel='Personeelsaantallen',
|
||
eenheid='Personeelsleden', kort='pers.')
|
||
|
||
|
||
def _huidig(query=''):
|
||
per = {r['School_ID']: r for r in central_query(
|
||
"SELECT School_ID, Aantal, Detail, Gesynchroniseerd FROM Personeelsaantal_huidig")}
|
||
rows = []
|
||
for s in alle_scholen():
|
||
if query and query.lower() not in s.naam.lower():
|
||
continue
|
||
h = per.get(s.id) or {}
|
||
rows.append({'Code': s.code, 'Naam_school': s.naam, 'Niveau': _niveau(s),
|
||
'Huidig': h['Aantal'] if h else '–', 'Laatste_synchronisatie': h.get('Gesynchroniseerd'),
|
||
'Informat': s.heeft(*INFORMAT_VELDEN), 'Campussen': _detail(h.get('Detail'))})
|
||
return rows
|
||
|
||
|
||
def _telling(peildatum, query=''):
|
||
per = {r['School_ID']: r for r in central_query(
|
||
"SELECT School_ID, Aantal, Detail, Schooljaar, Opgevraagd_op, Door FROM Personeelsaantal_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 Personeelsaantal_tellingen GROUP BY Peildatum ORDER BY Peildatum DESC")
|
||
|
||
|
||
def _tel(school, peildatum):
|
||
"""(ok, aantal of foutmelding, detail) voor één school."""
|
||
try:
|
||
with in_school(school):
|
||
totaal, detail = GoProService().tel(peildatum)
|
||
return True, totaal, detail
|
||
except Exception as e:
|
||
return False, str(e), None
|
||
|
||
|
||
@personeelsaantal_bp.route("/", methods=["GET"])
|
||
@login_required
|
||
@recht_vereist('personeelsaantal')
|
||
def personeelsaantal_page():
|
||
query = (request.values.get("query") or "").strip()
|
||
results = _huidig(query)
|
||
peildatum = _datum(request.args.get('peildatum'))
|
||
telling = _telling(peildatum, query) if peildatum else None
|
||
return render_template("leerlingenaantal.html", results=results, query=query, totaal=_som(results, 'Huidig'),
|
||
groepen=_groepen(results, 'Huidig'),
|
||
telling_groepen=_groepen(telling, 'Aantal') if telling else [],
|
||
peildatum=peildatum, telling=telling,
|
||
telling_totaal=_som(telling, 'Aantal') if telling else 0,
|
||
peildata=_peildata(), vandaag=date.today(), **WEERGAVE)
|
||
|
||
|
||
@personeelsaantal_bp.route("/sync-aanvragen", methods=["POST"])
|
||
@login_required
|
||
@recht_vereist('personeelsaantal')
|
||
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):
|
||
if not request.form.get("alle"):
|
||
flash(f"{school.naam}: Informat is nog niet ingesteld.", "error")
|
||
continue
|
||
ok, res, detail = _tel(school, date.today())
|
||
if not ok:
|
||
flash(f"{school.naam}: {res}", "error")
|
||
continue
|
||
central_query(
|
||
"INSERT INTO Personeelsaantal_huidig (School_ID, Aantal, Detail, Gesynchroniseerd) VALUES (%s, %s, %s, NOW()) "
|
||
"ON DUPLICATE KEY UPDATE Aantal = VALUES(Aantal), Detail = VALUES(Detail), Gesynchroniseerd = NOW()",
|
||
(school.id, res, json.dumps(detail) if detail else None), commit=True)
|
||
flash(f"{school.naam}: {res} personeelsleden" + (" (" + ", ".join(f"{d['naam']}: {d['aantal']}" for d in detail)
|
||
+ ")" if detail else "") + ".", "success")
|
||
return redirect(url_for('personeelsaantal.personeelsaantal_page'))
|
||
|
||
|
||
@personeelsaantal_bp.route("/peildatum", methods=["POST"])
|
||
@login_required
|
||
@recht_vereist('personeelsaantal')
|
||
def tel_peildatum():
|
||
peildatum = _datum(request.form.get('peildatum'))
|
||
if not peildatum:
|
||
flash("Kies een geldige datum.", "error")
|
||
return redirect(url_for('personeelsaantal.personeelsaantal_page'))
|
||
code = request.form.get('school_code')
|
||
scholen = [school_op_code(code)] if code else alle_scholen()
|
||
gelukt = 0
|
||
for school in scholen:
|
||
if not school or not school.heeft(*INFORMAT_VELDEN):
|
||
if school and code:
|
||
flash(f"{school.naam}: Informat is nog niet ingesteld.", "error")
|
||
continue
|
||
ok, res, detail = _tel(school, peildatum)
|
||
if not ok:
|
||
flash(f"{school.naam}: {res}", "error")
|
||
continue
|
||
central_query(
|
||
"INSERT INTO Personeelsaantal_tellingen (School_ID, Peildatum, Schooljaar, Aantal, Detail, Opgevraagd_op, Door) "
|
||
"VALUES (%s, %s, %s, %s, %s, NOW(), %s) ON DUPLICATE KEY UPDATE Schooljaar = VALUES(Schooljaar), "
|
||
"Aantal = VALUES(Aantal), Detail = VALUES(Detail), Opgevraagd_op = NOW(), Door = VALUES(Door)",
|
||
(school.id, peildatum, _automatisch_schooljaar(peildatum), res, json.dumps(detail) if detail else None,
|
||
session.get('user')), commit=True)
|
||
gelukt += 1
|
||
if gelukt:
|
||
flash(f"Personeel geteld op {peildatum:%d-%m-%Y} voor {gelukt} school/scholen.", "success")
|
||
return redirect(url_for('personeelsaantal.personeelsaantal_page', peildatum=peildatum.isoformat()))
|
||
|
||
|
||
@personeelsaantal_bp.route("/peildatum/verwijderen", methods=["POST"])
|
||
@login_required
|
||
@recht_vereist('personeelsaantal')
|
||
def verwijder_peildatum():
|
||
peildatum = _datum(request.form.get('peildatum'))
|
||
if peildatum:
|
||
central_query("DELETE FROM Personeelsaantal_tellingen WHERE Peildatum = %s", (peildatum,), commit=True)
|
||
flash(f"Telling van {peildatum:%d-%m-%Y} verwijderd.", "success")
|
||
return redirect(url_for('personeelsaantal.personeelsaantal_page'))
|
||
|
||
|
||
def _csv(naam, kop, rijen):
|
||
buf = io.StringIO()
|
||
w = csv.writer(buf, delimiter=';')
|
||
w.writerow(kop)
|
||
w.writerows(rijen)
|
||
return Response('' + buf.getvalue(), mimetype='text/csv',
|
||
headers={'Content-Disposition': f'attachment; filename={naam}'})
|
||
|
||
|
||
@personeelsaantal_bp.route("/export.csv", methods=["GET"])
|
||
@login_required
|
||
@recht_vereist('personeelsaantal')
|
||
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 Personeelsaantal_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, 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"personeelsaantallen_alle_peildata_{date.today():%Y%m%d}.csv",
|
||
['Onderwijsniveau', 'School', 'Instelling'] + [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"personeelsaantallen_{peildatum:%Y%m%d}.csv",
|
||
['Onderwijsniveau', 'School', 'Instelling', f"Personeelsleden op {peildatum:%d-%m-%Y}", 'Schooljaar',
|
||
'Opgevraagd op', 'Door'], rijen)
|
||
|
||
rows = _huidig(query)
|
||
rijen = []
|
||
for g in _groepen(rows, 'Huidig'):
|
||
for r in g['rijen']:
|
||
rijen.append([g['label'], r['Naam_school'], '', r['Huidig'] if isinstance(r['Huidig'], 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, 'Huidig'), ''])
|
||
return _csv(f"personeelsaantallen_{date.today():%Y%m%d}.csv",
|
||
['Onderwijsniveau', 'School', 'Instelling', 'Personeelsleden', 'Laatste synchronisatie'], rijen)
|