187 lines
8.5 KiB
Python
187 lines
8.5 KiB
Python
"""
|
||
Basisonderwijs › Bingel: leerlingenlijst uit Informat (alle instellingsnummers/campussen van de school) voor de import
|
||
in Bingel. Kolommen: Klas, Voornaam, Familienaam, Wachtwoord (leeg), Klasnummer, Geslacht (M/V).
|
||
|
||
Informat-leerlingen-API:
|
||
GET <API-URL>/students?schoolYear=&refdate= → de leerlingen (naam, voornaam, geslacht, inschrijvingsId)
|
||
GET <API-URL>/registrations?schoolYear=&refdate= → de inschrijvingen, met de klas
|
||
Het klasnummer haalt PIVOT niet uit Informat: binnen elke klas nummert het alfabetisch (familienaam, voornaam).
|
||
Gekoppeld op inschrijvingsId (anders op pPersoon). De velden worden op verschillende plaatsen en namen gezocht
|
||
(zie KANDIDATEN), bij een lijst (bv. klassen in de loop van het jaar) eerst de lopende.
|
||
"""
|
||
import json
|
||
import time
|
||
from datetime import date
|
||
|
||
from config import Config
|
||
|
||
KANDIDATEN = {
|
||
'klas': ('klascode', 'klasCode', 'klas', 'klasnaam', 'klasNaam', 'className', 'class'),
|
||
'geslacht': ('geslacht', 'gender', 'sex'),
|
||
'voornaam': ('voornaam', 'firstName', 'roepnaam'),
|
||
'naam': ('naam', 'lastName', 'familienaam'),
|
||
}
|
||
CACHE_SECONDEN = 10 * 60
|
||
|
||
|
||
class BingelFout(Exception):
|
||
pass
|
||
|
||
|
||
def _datum(w):
|
||
try:
|
||
return date.fromisoformat(str(w)[:10])
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _lopend(item, vandaag):
|
||
"""Is deze inschrijving lopend op `vandaag` (begin/einde onder verschillende namen)?"""
|
||
if not isinstance(item, dict):
|
||
return True
|
||
begin = next((_datum(item[k]) for k in ('begindatum', 'beginDatum', 'startDate', 'van') if item.get(k)), None)
|
||
einde = next((_datum(item[k]) for k in ('einddatum', 'eindDatum', 'endDate', 'tot') if item.get(k)), None)
|
||
return (not begin or begin <= vandaag) and (not einde or einde >= vandaag)
|
||
|
||
|
||
def _zoek(obj, namen, vandaag, diepte=0):
|
||
"""Eerste niet-lege waarde voor een van `namen`: eerst op dit niveau, dan in (lopende) geneste lijsten/dicts."""
|
||
if not isinstance(obj, dict) or diepte > 3:
|
||
return None
|
||
for n in namen:
|
||
w = obj.get(n)
|
||
if isinstance(w, dict): # bv. "klas": {"code": "3A", "naam": "3de leerjaar A"}
|
||
w = w.get('code') or w.get('naam') or w.get('name')
|
||
if w not in (None, '') and not isinstance(w, (dict, list)):
|
||
return w
|
||
for w in obj.values():
|
||
kinderen = [x for x in w if isinstance(x, dict)] if isinstance(w, list) else ([w] if isinstance(w, dict) else [])
|
||
kinderen.sort(key=lambda x: not _lopend(x, vandaag)) # lopende inschrijving eerst
|
||
for k in kinderen:
|
||
gevonden = _zoek(k, namen, vandaag, diepte + 1)
|
||
if gevonden not in (None, ''):
|
||
return gevonden
|
||
return None
|
||
|
||
|
||
def _geslacht(w):
|
||
t = str(w or '').strip().upper()
|
||
if t in ('M', 'MAN', 'MANNELIJK', 'JONGEN', 'MALE', '1'):
|
||
return 'M'
|
||
if t in ('V', 'F', 'VROUW', 'VROUWELIJK', 'MEISJE', 'FEMALE', '2'):
|
||
return 'V'
|
||
return t[:1]
|
||
|
||
|
||
def _nummer(w):
|
||
try:
|
||
return int(str(w).strip())
|
||
except (TypeError, ValueError):
|
||
return str(w).strip() if w not in (None, '') else None
|
||
|
||
|
||
def _inschrijvingen(svc, nummer, refdate):
|
||
"""Inschrijvingen (met klas) van één instellingsnummer. Geeft (lijst, fout)."""
|
||
import requests
|
||
from datetime import datetime, timezone
|
||
token = svc.get_informat_token(nummer)
|
||
if not token:
|
||
return [], f"Informat-authenticatie mislukt ({nummer})"
|
||
headers = {'Authorization': f'BEARER {token}', 'InstituteNo': nummer, 'Api-Version': '1', 'Accept': 'application/json',
|
||
'Timestamp': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.000Z')}
|
||
url = f"{svc.api_base.strip().rstrip('/')}/registrations"
|
||
params = {'schoolYear': Config.SCHOOLYEAR, 'refdate': refdate}
|
||
try:
|
||
r = requests.get(url, params=params, headers=headers, timeout=60)
|
||
if r.status_code == 405:
|
||
r = requests.post(url, params=params, headers=headers, timeout=60)
|
||
except Exception as e:
|
||
return [], f"inschrijvingen niet bereikbaar: {e}"
|
||
if r.status_code == 404:
|
||
return [], None
|
||
if r.status_code != 200:
|
||
return [], f"inschrijvingen: HTTP {r.status_code} {r.text[:150]}"
|
||
data = r.json()
|
||
return (data if isinstance(data, list) else (data.get('registrations') or data.get('items') or data.get('data') or [])), None
|
||
|
||
|
||
def leerlingen(vernieuw=False):
|
||
"""[{'klas','voornaam','naam','klasnummer','geslacht'}] gesorteerd op klas en klasnummer (alfabetisch), plus een voorbeeld
|
||
van de ruwe velden van de eerste leerling (om te controleren). Bewaard voor 10 minuten."""
|
||
from tenancy import huidige_school
|
||
from services.sync_service import SyncService
|
||
school = huidige_school()
|
||
bewaard = school.cache.get('bingel')
|
||
if bewaard and not vernieuw and time.time() - bewaard[0] < CACHE_SECONDEN:
|
||
return bewaard[1], bewaard[2], bewaard[3]
|
||
if not school.heeft('INFORMAT_INST_NUMBER', 'INFORMAT_ID', 'INFORMAT_SECRET'):
|
||
raise BingelFout("Informat is nog niet ingesteld voor deze school (🔌 Koppelingen › Informat).")
|
||
svc, vandaag = SyncService(), date.today()
|
||
ruw, gezien, inschr, waarschuwing = [], set(), {}, None
|
||
for nummer, naam in (svc.campussen or [(svc.institute_no, '')]):
|
||
ok, lijst = svc.informat_lijst(vandaag.isoformat(), Config.SCHOOLYEAR, nummer)
|
||
if not ok:
|
||
raise BingelFout(f"{naam + ': ' if naam else ''}{lijst}")
|
||
reg, fout = _inschrijvingen(svc, nummer, vandaag.isoformat())
|
||
if fout:
|
||
waarschuwing = f"Klassen konden niet opgehaald worden ({fout})."
|
||
for ins in sorted(reg, key=lambda x: _lopend(x, vandaag)): # lopende inschrijving wint
|
||
for k in ('inschrijvingsId', 'pInschrijving', 'id'):
|
||
if ins.get(k) not in (None, ''):
|
||
inschr[('i', str(ins[k]))] = ins
|
||
if ins.get('pPersoon') not in (None, ''):
|
||
inschr[('p', str(ins['pPersoon']))] = ins
|
||
for st in lijst:
|
||
sleutel = str(st.get('pPersoon') or st.get('internNummer') or st.get('id') or id(st))
|
||
if sleutel not in gezien:
|
||
gezien.add(sleutel)
|
||
ruw.append(st)
|
||
uit = []
|
||
for st in ruw:
|
||
ins = inschr.get(('i', str(st.get('inschrijvingsId')))) or inschr.get(('p', str(st.get('pPersoon')))) or {}
|
||
uit.append({'klas': str(_zoek(ins, KANDIDATEN['klas'], vandaag) or _zoek(st, KANDIDATEN['klas'], vandaag) or '').strip(),
|
||
'voornaam': str(_zoek(st, KANDIDATEN['voornaam'], vandaag) or '').strip(),
|
||
'naam': str(_zoek(st, KANDIDATEN['naam'], vandaag) or '').strip(),
|
||
'klasnummer': None,
|
||
'geslacht': _geslacht(_zoek(st, KANDIDATEN['geslacht'], vandaag))})
|
||
# Klasnummer: alfabetisch binnen de klas (familienaam, dan voornaam)
|
||
uit.sort(key=lambda r: (not r['klas'], r['klas'].lower(), r['naam'].lower(), r['voornaam'].lower()))
|
||
vorige, nr = None, 0
|
||
for r in uit:
|
||
nr = nr + 1 if r['klas'] == vorige else 1
|
||
vorige = r['klas']
|
||
r['klasnummer'] = nr if r['klas'] else None
|
||
# Om na te kijken: de velden van een inschrijving en een voorbeeld van een klas-item (eerste niet-lege)
|
||
alle_ins = list({id(v): v for v in inschr.values()}.values())
|
||
met_klas = next((i for i in alle_ins if _klas_items(i, vandaag)), None)
|
||
klas_item = _klas_items(met_klas, vandaag)[0] if met_klas else None
|
||
voorbeeld = {'leerling': _velden(ruw[0]) if ruw else [],
|
||
'inschrijving': _velden(met_klas or (alle_ins[0] if alle_ins else {})),
|
||
'klas_item': json.dumps(klas_item, ensure_ascii=False, default=str)[:600] if klas_item else ''}
|
||
school.cache['bingel'] = (time.time(), uit, voorbeeld, waarschuwing)
|
||
return uit, voorbeeld, waarschuwing
|
||
|
||
|
||
def _klas_items(ins, vandaag):
|
||
"""De klas-items van een inschrijving (bv. inschrKlassen), lopende eerst."""
|
||
items = []
|
||
for k, v in (ins or {}).items():
|
||
if 'klas' in k.lower() and isinstance(v, list):
|
||
items += [x for x in v if isinstance(x, dict)]
|
||
return sorted(items, key=lambda x: not _lopend(x, vandaag))
|
||
|
||
|
||
def _velden(obj, prefix=''):
|
||
"""Namen van de velden van één leerling (met geneste velden als a.b), om te zien wat Informat teruggeeft."""
|
||
uit = []
|
||
if isinstance(obj, dict):
|
||
for k, v in obj.items():
|
||
pad = f"{prefix}{k}"
|
||
if isinstance(v, dict):
|
||
uit += _velden(v, pad + '.')
|
||
elif isinstance(v, list) and v and isinstance(v[0], dict):
|
||
uit += _velden(v[0], pad + '.')
|
||
else:
|
||
uit.append(pad)
|
||
return uit
|