""" 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/') 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('//bewerken', methods=['GET', 'POST']) @superbeheerder_required def bewerken(code): return _bewerk(code, mag_alles=True) @scholen_bp.route('//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('//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'))