684 lines
31 KiB
Python
684 lines
31 KiB
Python
"""
|
||
Scholenbeheer.
|
||
|
||
- Superbeheerders (administrators in Gebruikers, of config.ini › superbeheerders via Google):
|
||
scholen toevoegen, alle instellingen wijzigen, tussen alle scholen wisselen.
|
||
- Gebruikers met het recht 'koppelingen' in een school: de koppelingen van DIE school wijzigen
|
||
(wachtwoorden, Informat, Google, ...), maar niet de code, domeinen of database.
|
||
- Iedereen kan wisselen naar een school waarin hij rechten heeft.
|
||
"""
|
||
import functools
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
import requests
|
||
from flask import Blueprint, render_template, request, redirect, url_for, flash, session, g
|
||
|
||
import tenancy
|
||
from tenancy import VELDEN, VELD, GROEPEN, alle_scholen, school_op_code, in_school
|
||
from database import central_query, execute_query, get_central_connection, vergeet_pools
|
||
import rechten
|
||
|
||
scholen_bp = Blueprint('scholen', __name__)
|
||
|
||
CODE_RE = re.compile(r'^[a-z0-9_-]{2,32}$')
|
||
DB_RE = re.compile(r'^[A-Za-z0-9_-]{1,64}$')
|
||
# Velden die enkel een superbeheerder mag wijzigen
|
||
ENKEL_SUPER_GROEPEN = {'database'}
|
||
|
||
|
||
def superbeheerder_required(f):
|
||
@functools.wraps(f)
|
||
def wrapped(*a, **kw):
|
||
if not session.get('auth') or not session.get('superbeheerder'):
|
||
flash("Enkel voor superbeheerders van de scholengroep.", "error")
|
||
return redirect(url_for('auth.login'))
|
||
return f(*a, **kw)
|
||
return wrapped
|
||
|
||
|
||
# Koppelingen van de eigen school: recht 'koppelingen' in die school
|
||
schoolbeheerder_required = rechten.recht_vereist('koppelingen')
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Overzicht en wisselen
|
||
# ------------------------------------------------------------------
|
||
@scholen_bp.route('/')
|
||
@superbeheerder_required
|
||
def overzicht():
|
||
tenancy.herlaad(force=True)
|
||
scholen = alle_scholen(ook_inactief=True)
|
||
# Onderverdeeld per onderwijsniveau; scholen zonder (geldig) niveau komen in de laatste groep
|
||
niveaus = list(tenancy.ONDERWIJSNIVEAUS) + [('', 'Onderwijsniveau niet gekozen')]
|
||
codes = {c for c, _ in tenancy.ONDERWIJSNIVEAUS}
|
||
niveau = lambda sch: sch.waarden.get('ONDERWIJSNIVEAU') if sch.waarden.get('ONDERWIJSNIVEAU') in codes else ''
|
||
groepen = [{'code': code or 'geen', 'label': label, 'scholen': [x for x in scholen if niveau(x) == code]}
|
||
for code, label in niveaus]
|
||
tests = laatste_tests()
|
||
return render_template('scholen.html', scholen=scholen, groepen=[g for g in groepen if g['scholen']],
|
||
badges={x.code: koppeling_badges(x, tests.get(x.id, {})) for x in scholen})
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Resultaten van de koppelingstest (voor de badges in het overzicht)
|
||
# ------------------------------------------------------------------
|
||
# Badge in het overzicht → onderdelen uit test_school() die erbij horen
|
||
BADGES = [
|
||
('Smartschool', ('SS_API_PLATFORM', 'SS_API_PW'), ('Smartschool webservices', 'Smartschool OneRoster')),
|
||
('Informat', ('INFORMAT_INST_NUMBER', 'INFORMAT_ID', 'INFORMAT_SECRET'), ('Informat',)),
|
||
('Google', None, ('Google Workspace',)),
|
||
('Signpost', ('M4S_CLIENT_ID', 'M4S_CLIENT_SECRET'), ('Signpost / M4S',)),
|
||
]
|
||
|
||
|
||
def _informat_ingesteld(school):
|
||
"""Informat telt als ingesteld met client-ID/secret en een instellingsnummer van de school óf van het internaat."""
|
||
return school.heeft('INFORMAT_ID', 'INFORMAT_SECRET') and (
|
||
school.heeft('INFORMAT_INST_NUMBER') or school.heeft('INFORMAT_INTERNAAT_NUMMER'))
|
||
|
||
|
||
def bewaar_test(school, resultaten):
|
||
for onderdeel, ok, detail in resultaten:
|
||
central_query(
|
||
"INSERT INTO Koppeling_tests (School_ID, Onderdeel, Ok, Detail, Getest_op) VALUES (%s, %s, %s, %s, NOW()) "
|
||
"ON DUPLICATE KEY UPDATE Ok = VALUES(Ok), Detail = VALUES(Detail), Getest_op = NOW()",
|
||
(school.id, onderdeel, None if ok is None else int(bool(ok)), (detail or '')[:500]), commit=True)
|
||
|
||
|
||
def laatste_tests():
|
||
"""{school_id: {onderdeel: rij}}"""
|
||
uit = {}
|
||
try:
|
||
for r in central_query("SELECT School_ID, Onderdeel, Ok, Detail, Getest_op FROM Koppeling_tests") or []:
|
||
uit.setdefault(r['School_ID'], {})[r['Onderdeel']] = r
|
||
except Exception:
|
||
pass
|
||
return uit
|
||
|
||
|
||
def koppeling_badges(school, tests):
|
||
"""[{label, toestand: ok|fout|ongetest|niet, titel}] volgens de laatste test. Een test van vóór de laatste
|
||
wijziging aan de school telt niet meer (dan 'ongetest')."""
|
||
from datetime import datetime
|
||
try:
|
||
gewijzigd = datetime.fromisoformat(school.versie).replace(microsecond=0) if school.versie else None
|
||
except ValueError:
|
||
gewijzigd = None
|
||
uit = []
|
||
for label, velden, onderdelen in BADGES:
|
||
if label == 'Informat':
|
||
ingesteld = _informat_ingesteld(school)
|
||
else:
|
||
ingesteld = school.google_sleutel is not None if velden is None else school.heeft(*velden)
|
||
if not ingesteld:
|
||
uit.append({'label': label, 'toestand': 'niet', 'titel': 'niet ingesteld'})
|
||
continue
|
||
rijen = [tests[o] for o in onderdelen if o in tests and tests[o]['Ok'] is not None]
|
||
if not rijen or (gewijzigd and any(r['Getest_op'] < gewijzigd for r in rijen)):
|
||
uit.append({'label': label, 'toestand': 'ongetest',
|
||
'titel': 'ingesteld, maar nog niet getest sinds de laatste wijziging – klik op Testen'})
|
||
continue
|
||
fout = [r for r in rijen if not r['Ok']]
|
||
wanneer = max(r['Getest_op'] for r in rijen).strftime('%d-%m-%Y %H:%M')
|
||
if fout:
|
||
uit.append({'label': label, 'toestand': 'fout',
|
||
'titel': f"test mislukt op {wanneer}: " + ' · '.join(r['Detail'] for r in fout)[:300]})
|
||
else:
|
||
uit.append({'label': label, 'toestand': 'ok', 'titel': f"test geslaagd op {wanneer}"})
|
||
return uit
|
||
|
||
|
||
@scholen_bp.route('/wissel/<code>')
|
||
def wissel(code):
|
||
"""Naar een andere school gaan: superbeheerders overal, anderen waar ze rechten hebben."""
|
||
if not session.get('auth'):
|
||
return redirect(url_for('auth.login'))
|
||
school = school_op_code(code)
|
||
if not school:
|
||
flash("Onbekende of inactieve school.", "error")
|
||
return redirect(url_for('main.dashboard'))
|
||
if not session.get('superbeheerder'):
|
||
gid = session.get('gebruiker_id')
|
||
if not gid or school.id not in rechten.scholen_met_rechten(gid):
|
||
flash(f"Je hebt geen rechten in {school.naam}.", "error")
|
||
return redirect(url_for('main.dashboard'))
|
||
session['school'] = school.code
|
||
rechten.bewaar_laatste_school(session.get('gebruiker_id'), school.code) # volgende keer meteen deze school
|
||
flash(f"Je werkt nu in {school.naam}.", "success")
|
||
from blueprints.auth import _onthoud_school
|
||
return _onthoud_school(redirect(url_for('main.dashboard')), school)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Formulier
|
||
# ------------------------------------------------------------------
|
||
# Velden die een uur van de dag zijn (0–23)
|
||
UUR_VELDEN = {'AFWEZIGHEID_SYNC_UUR'}
|
||
|
||
|
||
def _controleer_waarden(waarden):
|
||
"""Geeft een foutmelding als een getal-veld ongeldig is, anders None."""
|
||
for v in VELDEN:
|
||
w = str(waarden.get(v.key) or '').strip()
|
||
if v.soort != 'getal' or not w:
|
||
continue
|
||
if not w.isdigit():
|
||
return f"'{v.label}' moet een getal zijn (nu: {w})."
|
||
if v.key in UUR_VELDEN and not 0 <= int(w) <= 23:
|
||
return f"'{v.label}' moet een uur tussen 0 en 23 zijn (nu: {w})."
|
||
return None
|
||
|
||
|
||
def _form_waarden(school, mag_alles):
|
||
"""Leest het formulier. Lege geheimen = ongewijzigd laten (tenzij 'wissen' aangevinkt)."""
|
||
oud = dict(school.waarden) if school else {}
|
||
nieuw = dict(oud)
|
||
for v in VELDEN:
|
||
if v.groep in ENKEL_SUPER_GROEPEN and not mag_alles:
|
||
continue
|
||
naam = f"v_{v.key}"
|
||
if v.soort == 'janee':
|
||
nieuw[v.key] = 'ja' if request.form.get(naam) else 'nee'
|
||
continue
|
||
waarde = (request.form.get(naam) or '').strip()
|
||
if v.soort == 'geheim':
|
||
if request.form.get(f"wis_{v.key}"):
|
||
nieuw.pop(v.key, None)
|
||
elif waarde:
|
||
nieuw[v.key] = waarde
|
||
continue
|
||
if waarde:
|
||
nieuw[v.key] = waarde
|
||
else:
|
||
nieuw.pop(v.key, None)
|
||
return nieuw
|
||
|
||
|
||
def _google_sleutel_uit_form():
|
||
"""Geeft (sleutel_dict | None, foutmelding | None)."""
|
||
bestand = request.files.get('google_sleutel')
|
||
tekst = ''
|
||
if bestand and bestand.filename:
|
||
tekst = bestand.read().decode('utf-8', errors='replace')
|
||
elif (request.form.get('google_sleutel_tekst') or '').strip():
|
||
tekst = request.form['google_sleutel_tekst'].strip()
|
||
if not tekst:
|
||
return None, None
|
||
try:
|
||
data = json.loads(tekst)
|
||
except ValueError:
|
||
return None, "De Google-sleutel is geen geldig JSON-bestand."
|
||
if data.get('type') != 'service_account' or not data.get('private_key') or not data.get('client_email'):
|
||
return None, "Dit is geen sleutel van een Google-serviceaccount (type 'service_account')."
|
||
return data, None
|
||
|
||
|
||
def database_status(school):
|
||
"""'ok' (PIVOT-tabellen aanwezig), 'leeg' (database bestaat, tabellen ontbreken) of 'ontbreekt'."""
|
||
import mysql.connector
|
||
cfg = school.db_config()
|
||
from config import Config
|
||
try:
|
||
conn = mysql.connector.connect(host=cfg['host'], port=Config.DB_PORT, user=cfg['user'],
|
||
password=cfg['password'], connection_timeout=5)
|
||
except Exception:
|
||
return 'onbekend'
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = %s", (school.db_naam,))
|
||
if not cur.fetchone()[0]:
|
||
return 'ontbreekt'
|
||
cur.execute("SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s "
|
||
"AND LOWER(TABLE_NAME) = 'toestellen'", (school.db_naam,))
|
||
return 'ok' if cur.fetchone()[0] else 'leeg'
|
||
except Exception:
|
||
return 'onbekend'
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _render_form(school, nieuw, mag_alles, form=None):
|
||
return render_template('school_bewerken.html', school=school, nieuw=nieuw, mag_alles=mag_alles,
|
||
db_status=database_status(school) if school else None,
|
||
groepen=GROEPEN, velden=VELDEN, enkel_super=ENKEL_SUPER_GROEPEN,
|
||
form=form or {}, andere_scholen=[s for s in alle_scholen(ook_inactief=True) if not school or s.code != school.code])
|
||
|
||
|
||
@scholen_bp.route('/nieuw', methods=['GET', 'POST'])
|
||
@superbeheerder_required
|
||
def nieuw():
|
||
if request.method == 'GET':
|
||
return _render_form(None, True, True)
|
||
|
||
code = (request.form.get('code') or '').strip().lower()
|
||
naam = (request.form.get('naam') or '').strip()
|
||
db_naam = (request.form.get('db_naam') or '').strip()
|
||
fouten = []
|
||
if not CODE_RE.match(code):
|
||
fouten.append("De code mag enkel kleine letters, cijfers, - en _ bevatten (2-32 tekens).")
|
||
elif school_op_code(code, ook_inactief=True):
|
||
fouten.append(f"Er bestaat al een school met code '{code}'.")
|
||
if not naam:
|
||
fouten.append("Vul de naam van de school in.")
|
||
if not DB_RE.match(db_naam):
|
||
fouten.append("Ongeldige databasenaam.")
|
||
domeinen, fout = _domeinen(request.form.get('domeinen'))
|
||
if fout:
|
||
fouten.append(fout)
|
||
sleutel, fout = _google_sleutel_uit_form()
|
||
if fout:
|
||
fouten.append(fout)
|
||
if fouten:
|
||
for f in fouten:
|
||
flash(f, "error")
|
||
return _render_form(None, True, True, form=request.form)
|
||
|
||
waarden = _form_waarden(None, True)
|
||
fout = _controleer_waarden(waarden)
|
||
if fout:
|
||
flash(fout, "error")
|
||
return _render_form(None, True, True, form=request.form)
|
||
|
||
# Optioneel: database aanmaken (standaardstructuur, of een volledige kopie van een bestaande school)
|
||
keuze = request.form.get('kopieer_schema_van')
|
||
bron = school_op_code(keuze, ook_inactief=True) if keuze and keuze != STANDAARD else None
|
||
if keuze == STANDAARD:
|
||
ok, msg = maak_schooldatabase(db_naam, waarden)
|
||
flash(msg, "success" if ok else "error")
|
||
if not ok:
|
||
return _render_form(None, True, True, form=request.form)
|
||
elif bron:
|
||
if request.form.get('kopie_koppelingen'):
|
||
# Koppelingen en instellingen van de bron overnemen (database-server blijft dezelfde);
|
||
# aanpassen kan meteen daarna op de pagina Bewerken.
|
||
waarden = dict(bron.waarden)
|
||
sleutel = sleutel or bron.google_sleutel
|
||
ok, msg = kopieer_school_db(bron, db_naam, waarden)
|
||
flash(msg, "success" if ok else "error")
|
||
if not ok:
|
||
return _render_form(None, True, True, form=request.form)
|
||
|
||
tenancy.opslaan(code, naam, ','.join(domeinen), _hostnamen(request.form.get('hostnamen')), db_naam,
|
||
bool(request.form.get('actief')), waarden, sleutel, gebruiker=session.get('user'), nieuw=True)
|
||
flash(f"School {naam} toegevoegd.", "success")
|
||
if bron:
|
||
if request.form.get('kopie_koppelingen'):
|
||
flash(f"Koppelingen en instellingen van {bron.naam} overgenomen. Kijk ze hieronder na"
|
||
+ (" (de school staat actief: geplande synchronisaties lopen ook voor de kopie)."
|
||
if request.form.get('actief') else "."), "info")
|
||
if request.form.get('kopie_rechten'):
|
||
n = kopieer_rechten(bron, code, session.get('user'))
|
||
flash(f"Rechten van {n} gebruiker(s) overgenomen van {bron.naam}.", "success")
|
||
return redirect(url_for('scholen.bewerken', code=code))
|
||
|
||
|
||
@scholen_bp.route('/<code>/bewerken', methods=['GET', 'POST'])
|
||
@superbeheerder_required
|
||
def bewerken(code):
|
||
return _bewerk(code, mag_alles=True)
|
||
|
||
|
||
@scholen_bp.route('/<code>/database-aanmaken', methods=['POST'])
|
||
@superbeheerder_required
|
||
def database_aanmaken(code):
|
||
"""Database van een bestaande school (alsnog) aanmaken met de standaardstructuur.
|
||
Veilig om te herhalen: bestaande tabellen en gegevens blijven behouden."""
|
||
school = school_op_code(code, ook_inactief=True)
|
||
if not school:
|
||
flash("Onbekende school.", "error")
|
||
return redirect(url_for('scholen.overzicht'))
|
||
ok, msg = maak_schooldatabase(school.db_naam, school.waarden)
|
||
vergeet_pools(school.db_naam)
|
||
flash(msg, "success" if ok else "error")
|
||
return redirect(url_for('scholen.bewerken', code=school.code))
|
||
|
||
|
||
@scholen_bp.route('/mijn-school', methods=['GET', 'POST'])
|
||
@schoolbeheerder_required
|
||
def mijn_school():
|
||
return _bewerk(session['school'], mag_alles=bool(session.get('superbeheerder')))
|
||
|
||
|
||
def _bewerk(code, mag_alles):
|
||
school = school_op_code(code, ook_inactief=True)
|
||
if not school:
|
||
flash("Onbekende school.", "error")
|
||
return redirect(url_for('main.dashboard'))
|
||
if request.method == 'GET':
|
||
return _render_form(school, False, mag_alles)
|
||
|
||
if request.form.get('wis_google_sleutel'):
|
||
tenancy.wis_google_sleutel(school.code, session.get('user'))
|
||
school = school_op_code(code, ook_inactief=True)
|
||
|
||
sleutel, fout = _google_sleutel_uit_form()
|
||
if fout:
|
||
flash(fout, "error")
|
||
return _render_form(school, False, mag_alles, form=request.form)
|
||
|
||
if mag_alles:
|
||
naam = (request.form.get('naam') or '').strip() or school.naam
|
||
domeinen, fout = _domeinen(request.form.get('domeinen'), school.code)
|
||
db_naam = (request.form.get('db_naam') or '').strip()
|
||
if fout or not DB_RE.match(db_naam):
|
||
flash(fout or "Ongeldige databasenaam.", "error")
|
||
return _render_form(school, False, mag_alles, form=request.form)
|
||
hostnamen = _hostnamen(request.form.get('hostnamen'))
|
||
actief = bool(request.form.get('actief'))
|
||
else:
|
||
naam, domeinen, db_naam = school.naam, school.domeinen, school.db_naam
|
||
hostnamen, actief = ','.join(school.hostnamen), school.actief
|
||
|
||
waarden = _form_waarden(school, mag_alles)
|
||
fout = _controleer_waarden(waarden)
|
||
if fout:
|
||
flash(fout, "error")
|
||
return _render_form(school, False, mag_alles, form=request.form)
|
||
tenancy.opslaan(school.code, naam, ','.join(domeinen), hostnamen, db_naam, actief, waarden, sleutel,
|
||
gebruiker=session.get('user'))
|
||
vergeet_pools()
|
||
flash(f"Instellingen van {naam} opgeslagen.", "success")
|
||
if request.endpoint == 'scholen.bewerken':
|
||
return redirect(url_for('scholen.bewerken', code=school.code))
|
||
return redirect(url_for('scholen.mijn_school'))
|
||
|
||
|
||
def _domeinen(tekst, eigen_code=None):
|
||
domeinen = [d.strip().lower().lstrip('@') for d in re.split(r'[,;\s]+', tekst or '') if d.strip()]
|
||
if not domeinen:
|
||
return [], "Geef minstens één e-maildomein op (bv. school.be)."
|
||
for d in domeinen:
|
||
if not re.match(r'^[a-z0-9.-]+\.[a-z]{2,}$', d):
|
||
return [], f"'{d}' is geen geldig domein."
|
||
# Meerdere scholen mogen hetzelfde domein delen (bv. basis- en secundaire school van één campus).
|
||
# Wie zich met Google aanmeldt, komt dan in de school waar hij toegang toe heeft (zie auth.callback).
|
||
return domeinen, None
|
||
|
||
|
||
def _hostnamen(tekst):
|
||
return ','.join(h.strip().lower() for h in re.split(r'[,;\s]+', tekst or '') if h.strip())
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Database aanmaken
|
||
# ------------------------------------------------------------------
|
||
STANDAARD = '__standaard__'
|
||
SCHOOL_SQL = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'sql', 'school.sql')
|
||
|
||
|
||
def sql_statements(pad):
|
||
"""Leest een .sql-bestand en geeft de afzonderlijke statements (commentaar weggelaten)."""
|
||
with open(pad, encoding='utf-8') as f:
|
||
regels = [r for r in f.read().split('\n') if not r.strip().startswith('--')]
|
||
return [st.strip() for st in re.split(r';\s*\n', '\n'.join(regels) + '\n') if st.strip()]
|
||
|
||
|
||
def maak_schooldatabase(db_naam, waarden=None):
|
||
"""Maakt db_naam aan met de standaardstructuur uit sql/school.sql (lege tabellen)."""
|
||
import mysql.connector
|
||
from config import Config
|
||
waarden = waarden or {}
|
||
if not DB_RE.match(db_naam or ''):
|
||
return False, "Ongeldige databasenaam."
|
||
try:
|
||
conn = mysql.connector.connect(
|
||
host=waarden.get('DB_HOST') or Config.DB_HOST, port=Config.DB_PORT,
|
||
user=waarden.get('DB_USER') or Config.DB_USER,
|
||
password=waarden.get('DB_PASSWORD') or Config.DB_PASSWORD, connection_timeout=10,
|
||
charset='utf8mb4', collation='utf8mb4_unicode_ci')
|
||
except Exception as e:
|
||
return False, f"Kan niet verbinden met de databaseserver: {e}"
|
||
cur = conn.cursor()
|
||
try:
|
||
cur.execute(f"CREATE DATABASE IF NOT EXISTS `{db_naam}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
|
||
cur.execute(f"USE `{db_naam}`")
|
||
stmts = sql_statements(SCHOOL_SQL)
|
||
for st in stmts:
|
||
cur.execute(st)
|
||
conn.commit()
|
||
n = sum(1 for st in stmts if st.upper().startswith('CREATE TABLE'))
|
||
return True, f"Database {db_naam} aangemaakt met de standaardstructuur ({n} tabellen)."
|
||
except Exception as e:
|
||
conn.rollback()
|
||
return False, f"Database aanmaken mislukt: {e}"
|
||
finally:
|
||
cur.close()
|
||
conn.close()
|
||
def kopieer_school_db(bron, db_naam, waarden):
|
||
"""Maakt db_naam aan (zelfde server) als volledige kopie van de database van `bron`: alle tabellen mét gegevens.
|
||
De doeldatabase moet nieuw of leeg zijn. Mislukt het, dan wordt een database die we zelf aanmaakten weer verwijderd."""
|
||
bron_db = bron.db_naam
|
||
if waarden.get('DB_HOST') and waarden.get('DB_HOST') != (bron.waarden.get('DB_HOST') or ''):
|
||
return False, "Een school kopiëren kan enkel binnen dezelfde databaseserver."
|
||
if db_naam == bron_db:
|
||
return False, "Kies een andere databasenaam dan die van de school die je kopieert."
|
||
conn = get_central_connection()
|
||
cur = conn.cursor()
|
||
zelf_aangemaakt = False
|
||
try:
|
||
cur.execute("SELECT 1 FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = %s", (db_naam,))
|
||
bestond = bool(cur.fetchall())
|
||
if bestond:
|
||
cur.execute(f"SHOW FULL TABLES FROM `{db_naam}` WHERE Table_type = 'BASE TABLE'")
|
||
if cur.fetchall():
|
||
return False, (f"De database {db_naam} bestaat al en bevat tabellen. Kies een nieuwe databasenaam "
|
||
"om er een kopie in te zetten.")
|
||
else:
|
||
cur.execute(f"CREATE DATABASE `{db_naam}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
|
||
zelf_aangemaakt = True
|
||
cur.execute(f"SHOW FULL TABLES FROM `{bron_db}` WHERE Table_type = 'BASE TABLE'")
|
||
tabellen = [r[0] for r in cur.fetchall()]
|
||
cur.execute("SET FOREIGN_KEY_CHECKS = 0")
|
||
rijen = 0
|
||
for t in tabellen:
|
||
cur.execute(f"CREATE TABLE `{db_naam}`.`{t}` LIKE `{bron_db}`.`{t}`")
|
||
cur.execute(f"INSERT INTO `{db_naam}`.`{t}` SELECT * FROM `{bron_db}`.`{t}`")
|
||
rijen += max(cur.rowcount or 0, 0)
|
||
cur.execute("SET FOREIGN_KEY_CHECKS = 1")
|
||
conn.commit()
|
||
return True, (f"Database {db_naam} aangemaakt als kopie van {bron.naam}: "
|
||
f"{len(tabellen)} tabellen, {rijen} rijen gegevens.")
|
||
except Exception as e:
|
||
try:
|
||
conn.rollback()
|
||
cur.execute("SET FOREIGN_KEY_CHECKS = 1")
|
||
if zelf_aangemaakt:
|
||
cur.execute(f"DROP DATABASE IF EXISTS `{db_naam}`")
|
||
except Exception:
|
||
pass
|
||
return False, f"School kopiëren mislukt: {e}"
|
||
finally:
|
||
cur.close()
|
||
conn.close()
|
||
|
||
|
||
def kopieer_rechten(bron, nieuwe_code, door=None):
|
||
"""Geeft iedereen die rechten heeft in `bron` dezelfde rechten in de nieuwe school."""
|
||
nieuw = school_op_code(nieuwe_code, ook_inactief=True)
|
||
if not nieuw:
|
||
return 0
|
||
central_query("INSERT IGNORE INTO Gebruiker_rechten (Gebruiker_ID, School_ID, Onderdeel_ID, Toegekend_op, Toegekend_door) "
|
||
"SELECT Gebruiker_ID, %s, Onderdeel_ID, NOW(), %s FROM Gebruiker_rechten WHERE School_ID = %s",
|
||
(nieuw.id, door, bron.id), commit=True)
|
||
return central_query("SELECT COUNT(DISTINCT Gebruiker_ID) AS n FROM Gebruiker_rechten WHERE School_ID = %s",
|
||
(nieuw.id,))[0]['n']
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Verbindingstest
|
||
# ------------------------------------------------------------------
|
||
def test_school(school):
|
||
"""Test elke ingestelde koppeling. Geeft een lijst (onderdeel, ok|None, detail)."""
|
||
res = []
|
||
with in_school(school):
|
||
# Database (eerst tabelnamen met hoofdletters herstellen, voor een database die van Windows komt)
|
||
try:
|
||
from services.tabelnamen import herstel_school
|
||
for m in herstel_school(school):
|
||
res.append(('Tabelnamen', 'mislukt' not in m and 'samen' not in m, m))
|
||
except Exception as e:
|
||
res.append(('Tabelnamen', False, str(e)[:200]))
|
||
try:
|
||
n = execute_query("SELECT COUNT(*) AS n FROM information_schema.tables WHERE table_schema = DATABASE()")[0]['n']
|
||
heeft = execute_query("SHOW TABLES LIKE 'Toestellen'")
|
||
res.append(('Database', bool(heeft), f"{school.db_naam}: {n} tabellen"
|
||
+ ('' if heeft else " – tabel Toestellen ontbreekt (python beheer.py nieuwe-schooldatabase)")))
|
||
except Exception as e:
|
||
res.append(('Database', False, str(e)))
|
||
|
||
# Smartschool webservices
|
||
if school.heeft('SS_API_PLATFORM', 'SS_API_PW'):
|
||
try:
|
||
from services.smartschool_api import SmartschoolService
|
||
from config import Config
|
||
ss = SmartschoolService(Config.SS_API_PLATFORM, Config.SS_API_PW)
|
||
raw = ss.client.service.getClassListJson(Config.SS_API_PW)
|
||
if isinstance(raw, (int, float)) or (isinstance(raw, str) and raw.strip().lstrip('-').isdigit()):
|
||
res.append(('Smartschool webservices', False, f"foutcode {raw} (wachtwoord of rechten)"))
|
||
else:
|
||
res.append(('Smartschool webservices', True, Config.SS_API_PLATFORM))
|
||
except Exception as e:
|
||
res.append(('Smartschool webservices', False, str(e)[:200]))
|
||
else:
|
||
res.append(('Smartschool webservices', None, 'niet ingesteld'))
|
||
|
||
# Smartschool OneRoster
|
||
if school.heeft('SS_API_PLATFORM', 'SS_CLIENT_ID', 'SS_CLIENT_SECRET'):
|
||
try:
|
||
from config import Config
|
||
r = requests.post(Config.SS_TOKEN_URL, data={"grant_type": "client_credentials"},
|
||
auth=(Config.SS_CLIENT_ID, Config.SS_CLIENT_SECRET), timeout=10)
|
||
res.append(('Smartschool OneRoster', r.ok and bool(r.json().get('access_token')),
|
||
'token ontvangen' if r.ok else f"HTTP {r.status_code}"))
|
||
except Exception as e:
|
||
res.append(('Smartschool OneRoster', False, str(e)[:200]))
|
||
else:
|
||
res.append(('Smartschool OneRoster', None, 'niet ingesteld'))
|
||
|
||
# Informat: eerst de leerlingen-API; lukt die niet, dan de Internaat-API (een internaat gebruikt in
|
||
# Informat geen leerlingen maar internen, dus de leerlingen-scope wordt daar geweigerd)
|
||
if _informat_ingesteld(school):
|
||
fout = None
|
||
if school.heeft('INFORMAT_INST_NUMBER'):
|
||
try:
|
||
from services.sync_service import SyncService
|
||
if SyncService().get_informat_token():
|
||
res.append(('Informat', True, 'token ontvangen (leerlingen)'))
|
||
else:
|
||
fout = 'authenticatie mislukt'
|
||
except Exception as e:
|
||
fout = str(e)[:200]
|
||
else:
|
||
fout = 'geen instellingsnummer voor leerlingen'
|
||
if fout is not None:
|
||
try:
|
||
from services.internaat_service import InternaatService
|
||
svc = InternaatService()
|
||
if not svc.nummers:
|
||
raise RuntimeError('geen instellingsnummer')
|
||
for nr in svc.nummers:
|
||
svc._token(nr)
|
||
res.append(('Informat', True, f"token ontvangen (internaat {', '.join(svc.nummers)})"))
|
||
except Exception as e:
|
||
res.append(('Informat', False, f"{fout}; internaat: {str(e)[:150]}"))
|
||
else:
|
||
res.append(('Informat', None, 'niet ingesteld'))
|
||
|
||
# Google
|
||
if school.google_sleutel and not school.heeft('ADMIN_EMAIL'):
|
||
res.append(('Google Workspace', False, "Sleutel staat erin, maar 'Beheerder (Google-admin)' is nog niet "
|
||
"ingevuld (Algemeen). PIVOT moet weten namens welk beheerdersaccount het werkt."))
|
||
elif school.google_sleutel:
|
||
try:
|
||
from services.google_service import GoogleAdminService
|
||
svc = GoogleAdminService()
|
||
svc.service.users().list(customer='my_customer', maxResults=1).execute()
|
||
res.append(('Google Workspace', True, f"namens {school.get('ADMIN_EMAIL')}"))
|
||
except Exception as e:
|
||
res.append(('Google Workspace', False, google_fout_uitleg(str(e), school)))
|
||
else:
|
||
res.append(('Google Workspace', None, 'geen serviceaccount-sleutel'))
|
||
|
||
# Signpost
|
||
if school.heeft('M4S_CLIENT_ID', 'M4S_CLIENT_SECRET'):
|
||
from services.signpost_byod_service import get_signpost_token
|
||
ok = bool(get_signpost_token())
|
||
res.append(('Signpost / M4S', ok, 'token ontvangen' if ok else 'authenticatie mislukt'))
|
||
else:
|
||
res.append(('Signpost / M4S', None, 'niet ingesteld'))
|
||
return res
|
||
|
||
|
||
def google_fout_uitleg(fout, school):
|
||
"""Vertaalt de meest voorkomende Google-fouten naar wat je moet nakijken."""
|
||
admin = school.get('ADMIN_EMAIL')
|
||
f = fout.lower()
|
||
# Bij een HttpError staat de eigenlijke reden na 'returned "...": die tonen i.p.v. de lange URL
|
||
m = re.search(r'returned "([^"]+)"', fout)
|
||
reden = m.group(1) if m else fout
|
||
if 'has not been used' in f or 'is disabled' in f or 'accessnotconfigured' in f or 'service_disabled' in f:
|
||
uitleg = ("De Admin SDK API staat nog niet aan in het Google Cloud-project van het serviceaccount "
|
||
"(console.cloud.google.com › API's en services › Bibliotheek › Admin SDK API › Inschakelen).")
|
||
elif 'unauthorized_client' in f:
|
||
uitleg = ("De domeinbrede delegatie is (nog) niet in orde: controleer in admin.google.com › Beveiliging › "
|
||
"API-beheer › Domeinbrede delegatie de client-ID (Unieke ID van het serviceaccount) en de OAuth-bereiken. "
|
||
"Een nieuwe instelling kan tot een uur duren.")
|
||
elif 'invalid_grant' in f:
|
||
uitleg = (f"Google kent het beheerdersaccount '{admin}' niet of het is geschorst. "
|
||
"Vul bij 'Beheerder (Google-admin)' een bestaand account van dit domein in.")
|
||
elif 'not authorized' in f or '403' in f:
|
||
uitleg = (f"'{admin}' mag de gebruikerslijst niet lezen in Google. Controleer in admin.google.com › Account › "
|
||
"Beheerdersrollen dat dit account superbeheerder is (een gewone gebruiker of een beheerder met een "
|
||
"beperkte rol volstaat niet), en dat het in dezelfde Google-organisatie zit als de domeinbrede delegatie.")
|
||
elif 'invalid input' in f:
|
||
uitleg = (f"Google aanvaardt '{admin}' niet als beheerder van dit domein. Controleer of 'Beheerder (Google-admin)' "
|
||
"een echt beheerdersaccount is (geen groep of alias) van hetzelfde domein als het serviceaccount.")
|
||
else:
|
||
return fout[:300]
|
||
return f"{uitleg} (Google: {reden[:200]})"
|
||
|
||
|
||
@scholen_bp.route('/<code>/test', methods=['POST'])
|
||
def test(code):
|
||
if not session.get('auth'):
|
||
return redirect(url_for('auth.login'))
|
||
if not session.get('superbeheerder') and not rechten.heeft_recht('koppelingen'):
|
||
flash("Onvoldoende rechten voor deze actie.", "error")
|
||
return redirect(url_for('main.dashboard'))
|
||
if not session.get('superbeheerder') and code != session.get('school'):
|
||
flash("Je kan enkel je eigen school testen.", "error")
|
||
return redirect(url_for('main.dashboard'))
|
||
school = school_op_code(code, ook_inactief=True)
|
||
if not school:
|
||
flash("Onbekende school.", "error")
|
||
return redirect(url_for('main.dashboard'))
|
||
resultaten = test_school(school)
|
||
try:
|
||
bewaar_test(school, resultaten)
|
||
except Exception as e:
|
||
sys.stderr.write(f"PIVOT: testresultaat bewaren mislukt: {e}\n")
|
||
return render_template('school_test.html', school=school, resultaten=resultaten)
|
||
|
||
|
||
@scholen_bp.route('/alles-testen', methods=['POST'])
|
||
@superbeheerder_required
|
||
def alles_testen():
|
||
"""Test de koppelingen van alle actieve scholen en werkt de badges in het overzicht bij."""
|
||
goed, fout = 0, []
|
||
for school in alle_scholen():
|
||
try:
|
||
res = test_school(school)
|
||
bewaar_test(school, res)
|
||
if any(ok is False for _, ok, _ in res):
|
||
fout.append(school.naam)
|
||
else:
|
||
goed += 1
|
||
except Exception as e:
|
||
fout.append(f"{school.naam} ({e})")
|
||
flash(f"{goed} school/scholen zonder fouten." + (f" Met fouten: {', '.join(fout)}." if fout else ""),
|
||
"warning" if fout else "success")
|
||
return redirect(url_for('scholen.overzicht'))
|