234 lines
12 KiB
Python
234 lines
12 KiB
Python
"""
|
||
Internaat: de internen uit Informat (Internaat WEB API, versie 1).
|
||
|
||
- Token: zelfde identityserver en client-ID/secret als de andere Informat-API's, scope
|
||
api_informat_sas_internaat.<instellingsnummer van het internaat>
|
||
(meerdere instellingsnummers mogen in één token, gescheiden door een spatie).
|
||
- GET <API-URL>/pupils?schoolYear=2026-27&refdate=JJJJ-MM-DD → de internen met hun gegevens, relaties, ...
|
||
- GET <API-URL>/registrations?schoolYear=2026-27 → de inschrijvingen (internaat, vestiging, kamers, ...)
|
||
Headers: InstituteNo (instellingsnummer) en Api-Version: 1. 404 = geen gegevens.
|
||
Het resultaat wordt 10 minuten per school, schooljaar en peildatum bewaard.
|
||
"""
|
||
import time
|
||
from datetime import date
|
||
|
||
import requests
|
||
|
||
from config import Config
|
||
|
||
CACHE_SECONDEN = 600
|
||
TIMEOUT = (15, 180)
|
||
STANDAARD_API = 'https://internaatapi.informatsoftware.be'
|
||
STANDAARD_SCOPE = 'api_informat_sas_internaat'
|
||
|
||
|
||
class InternaatFout(Exception):
|
||
pass
|
||
|
||
|
||
def _datum(w):
|
||
if not w:
|
||
return None
|
||
try:
|
||
return date.fromisoformat(str(w)[:10])
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _lopend(item, peildatum):
|
||
"""Loopt dit (inschrijving of kamer) op de peildatum?"""
|
||
begin, einde = _datum(item.get('begindatum')), _datum(item.get('einddatum'))
|
||
return (not begin or begin <= peildatum) and (not einde or einde >= peildatum)
|
||
|
||
|
||
class InternaatService:
|
||
def __init__(self):
|
||
from tenancy import huidige_school, informat_campussen
|
||
self.school = huidige_school()
|
||
self.base = (self.school.get('INFORMAT_INTERNAAT_API') or STANDAARD_API).rstrip('/')
|
||
scope = (self.school.get('INFORMAT_INTERNAAT_SCOPE') or STANDAARD_SCOPE).strip()
|
||
self.scope_basis = scope.rsplit('.', 1)[0] if scope.rsplit('.', 1)[-1].isdigit() else scope
|
||
# Instellingsnummer(s) van het internaat; leeg = die van de Informat-koppeling van de school
|
||
eigen = informat_campussen(self.school.waarden.get('INFORMAT_INTERNAAT_NUMMER'))
|
||
self.nummers = eigen or informat_campussen(self.school.waarden.get('INFORMAT_INST_NUMBER'))
|
||
|
||
# ---------------- authenticatie ----------------
|
||
def _token(self, nummer):
|
||
sleutel = ('internaat_token', nummer)
|
||
bewaard = self.school.cache.get(sleutel)
|
||
if bewaard and time.time() < bewaard[0]:
|
||
return bewaard[1]
|
||
scope = f"{self.scope_basis}.{nummer}"
|
||
try:
|
||
r = requests.post(Config.INFORMAT_AUTH_URL, data={
|
||
'grant_type': 'client_credentials', 'client_id': Config.INFORMAT_ID,
|
||
'client_secret': Config.INFORMAT_SECRET, 'scope': scope}, timeout=15)
|
||
except Exception as e:
|
||
raise InternaatFout(f"Informat (identityserver) is niet bereikbaar: {e}")
|
||
try:
|
||
js = r.json()
|
||
except Exception:
|
||
js = {}
|
||
if r.status_code != 200 or not js.get('access_token'):
|
||
fout = js.get('error') or f"HTTP {r.status_code}"
|
||
if fout == 'invalid_scope':
|
||
raise InternaatFout(f"Informat weigert de scope {scope} (invalid_scope). Heeft het internaat met "
|
||
f"instellingsnummer {nummer} in Informat de app toegang gegeven tot de Internaat-API? "
|
||
"Klopt het instellingsnummer van het internaat (Koppelingen › Informat)?")
|
||
if fout == 'invalid_client':
|
||
raise InternaatFout("Informat kent deze client-ID/secret niet (invalid_client). Kijk Koppelingen › Informat na.")
|
||
raise InternaatFout(f"Aanmelden bij de Internaat-API mislukt: {fout}")
|
||
self.school.cache[sleutel] = (time.time() + int(js.get('expires_in') or 3600) - 60, js['access_token'])
|
||
return js['access_token']
|
||
|
||
def _get(self, pad, nummer, params):
|
||
headers = {'Authorization': f"BEARER {self._token(nummer)}", 'InstituteNo': nummer, 'Api-Version': '1',
|
||
'Accept': 'application/json'}
|
||
url = f"{self.base}{pad}"
|
||
try:
|
||
r = requests.get(url, params=params, headers=headers, timeout=TIMEOUT)
|
||
if r.status_code == 405: # de voorbeelden in de handleiding gebruiken POST
|
||
r = requests.post(url, params=params, headers=headers, timeout=TIMEOUT)
|
||
except requests.exceptions.Timeout:
|
||
raise InternaatFout("De Internaat-API van Informat antwoordde niet op tijd. Probeer later opnieuw.")
|
||
except Exception as e:
|
||
raise InternaatFout(f"De Internaat-API van Informat is niet bereikbaar: {e}")
|
||
if r.status_code == 404:
|
||
return [] # volgens Informat: geen gegevens
|
||
if r.status_code in (401, 403):
|
||
raise InternaatFout(f"Informat weigert de toegang tot de internen van {nummer} ({r.status_code}).")
|
||
if r.status_code != 200:
|
||
raise InternaatFout(f"Internaat-API ({nummer}): HTTP {r.status_code} {r.text[:200]}")
|
||
try:
|
||
data = r.json()
|
||
except Exception:
|
||
raise InternaatFout(f"Onverwacht antwoord van de Internaat-API: {r.text[:200]}")
|
||
return data if isinstance(data, list) else (data.get('items') or data.get('data') or [])
|
||
|
||
# ---------------- internen ----------------
|
||
def internen(self, schooljaar, peildatum=None, vernieuw=False):
|
||
"""(rijen, opgehaald) met één rij per intern die op de peildatum ingeschreven is."""
|
||
if not self.nummers:
|
||
raise InternaatFout("Er is nog geen instellingsnummer voor het internaat ingesteld (Koppelingen › Informat).")
|
||
peildatum = peildatum or date.today()
|
||
sleutel = ('internen', schooljaar, peildatum.isoformat())
|
||
bewaard = self.school.cache.get(sleutel)
|
||
if bewaard and not vernieuw and time.time() - bewaard[0] < CACHE_SECONDEN:
|
||
return bewaard[1], bewaard[0]
|
||
rijen, gezien = [], set()
|
||
for nummer, campusnaam in self.nummers:
|
||
inschrijvingen = {}
|
||
for ins in self._get('/registrations', nummer, {'schoolYear': schooljaar}):
|
||
if _lopend(ins, peildatum) or str(ins.get('pPersoon')) not in inschrijvingen:
|
||
inschrijvingen[str(ins.get('pPersoon'))] = ins
|
||
for p in self._get('/pupils', nummer, {'schoolYear': schooljaar, 'refdate': peildatum.isoformat()}):
|
||
sleutel_p = str(p.get('persoonId') or p.get('pPersoon'))
|
||
if sleutel_p in gezien: # zelfde intern via een tweede instellingsnummer
|
||
continue
|
||
gezien.add(sleutel_p)
|
||
ins = inschrijvingen.get(str(p.get('pPersoon'))) or {}
|
||
kamer = next((k for k in ins.get('kamers') or [] if _lopend(k, peildatum)), None)
|
||
ouders = []
|
||
for rel in sorted(p.get('relaties') or [], key=lambda x: (x.get('lpv') or 9)):
|
||
if rel.get('isOverleden'):
|
||
continue
|
||
gsm = next((c.get('nr') for c in rel.get('comnrs') or [] if (c.get('soort') or '').lower() == 'gsm'), None) \
|
||
or next((c.get('nr') for c in rel.get('comnrs') or []), None)
|
||
mail = next((e.get('email') for e in rel.get('emails') or [] if e.get('schoolcom')), None) \
|
||
or next((e.get('email') for e in rel.get('emails') or []), None)
|
||
ouders.append({'type': rel.get('type') or '', 'naam': f"{rel.get('voornaam') or ''} {rel.get('naam') or ''}".strip(),
|
||
'gsm': gsm or '', 'email': mail or ''})
|
||
adres = next((a for a in p.get('adressen') or [] if a.get('isVerblijf')), None) or \
|
||
next(iter(p.get('adressen') or []), None) or {}
|
||
rijen.append({
|
||
'id': p.get('pPersoon'),
|
||
'naam': (p.get('naam') or '').strip(), 'voornaam': (p.get('voornaam') or '').strip(),
|
||
'geboortedatum': _datum(p.get('geboortedatum')), 'geslacht': p.get('geslacht') or '',
|
||
'stamnummer': ins.get('stamnr') or '',
|
||
'internaat': ins.get('internaat') or campusnaam or nummer, 'nummer': nummer,
|
||
'vestiging': ins.get('vestiging') or '',
|
||
'begin': _datum(ins.get('begindatum')), 'einde': _datum(ins.get('einddatum')),
|
||
'financiering': ins.get('fin') or '',
|
||
'kamer': ' '.join(x for x in (
|
||
f"blok {kamer.get('blok')}" if kamer and kamer.get('blok') else '',
|
||
f"verd. {kamer.get('verdiep')}" if kamer and kamer.get('verdiep') not in (None, '') else '',
|
||
f"nr. {kamer.get('nr')}" if kamer and kamer.get('nr') not in (None, '') else '') if x),
|
||
'kamer_sort': f"{(kamer or {}).get('blok') or ''}-{str((kamer or {}).get('verdiep') or '').zfill(3)}-{str((kamer or {}).get('nr') or '').zfill(5)}" if kamer else '',
|
||
'ouders': ouders,
|
||
'gemeente': ' '.join(x for x in (adres.get('postcode'), adres.get('gemeente')) if x),
|
||
'huisdokter': ' – '.join(x for x in (p.get('huisdokter'), p.get('telefoonHuisdokter')) if x),
|
||
})
|
||
rijen.sort(key=lambda r: (r['naam'].lower(), r['voornaam'].lower()))
|
||
nu = time.time()
|
||
self.school.cache[sleutel] = (nu, rijen)
|
||
return rijen, nu
|
||
|
||
|
||
# ---------------------------------------------------------------------
|
||
# Bestaat de intern in Smartschool? (gebruikersnaam = "voornaam achternaam")
|
||
# ---------------------------------------------------------------------
|
||
SS_CACHE_SECONDEN = 3600
|
||
|
||
|
||
def _naamdeel(tekst):
|
||
"""'Van den Broeck' → 'vandenbroeck', 'Zoë' → 'zoe'."""
|
||
import unicodedata
|
||
t = unicodedata.normalize('NFKD', str(tekst or '')).encode('ascii', 'ignore').decode().lower()
|
||
return ''.join(c for c in t if c.isalnum() or c == '-')
|
||
|
||
|
||
def ss_kandidaten(voornaam, naam):
|
||
"""Mogelijke Smartschool-gebruikersnamen: eerst "voornaam achternaam" (met spatie), dan varianten."""
|
||
v, n = _naamdeel(voornaam), _naamdeel(naam)
|
||
vol = ' '.join(f"{str(voornaam).strip()} {str(naam).strip()}".split())
|
||
# Op het internaat: "voornaam achternaam" met een spatie (eerst zoals in Informat, dan in kleine letters)
|
||
uit = [vol, vol.lower(), f"{v} {n}",
|
||
f"{v}.{n}", f"{v}.{n.replace('-', '')}", f"{v.replace('-', '')}.{n.replace('-', '')}", f"{v}{n}"]
|
||
return [x for x in dict.fromkeys(uit) if x.strip('. ')]
|
||
|
||
|
||
def smartschool_status(personen):
|
||
"""personen: [{'pid', 'voornaam', 'naam'}] → {pid: gebruikersnaam of None}. 1 uur bewaard per school."""
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from tenancy import huidige_school, in_school
|
||
school = huidige_school()
|
||
if not school.heeft('SS_API_PLATFORM', 'SS_API_PW'):
|
||
raise InternaatFout("Smartschool is nog niet ingesteld voor deze school (🔌 Koppelingen › Smartschool).")
|
||
cache = school.cache.setdefault('ss_internen', {})
|
||
nu = time.time()
|
||
uit, te_doen = {}, []
|
||
for p in personen:
|
||
sleutel = (str(p.get('pid')), p.get('voornaam') or '', p.get('naam') or '')
|
||
hit = cache.get(sleutel)
|
||
if hit and nu - hit[0] < SS_CACHE_SECONDEN:
|
||
uit[sleutel[0]] = hit[1]
|
||
else:
|
||
te_doen.append(sleutel)
|
||
if not te_doen:
|
||
return uit
|
||
lokaal = threading.local()
|
||
|
||
def zoek(sleutel):
|
||
pid, voornaam, naam = sleutel
|
||
with in_school(school):
|
||
if not hasattr(lokaal, 'ss'):
|
||
from services.smartschool_api import SmartschoolService
|
||
lokaal.ss = SmartschoolService(Config.SS_API_PLATFORM, Config.SS_API_PW)
|
||
gevonden = None
|
||
for kandidaat in ss_kandidaten(voornaam, naam):
|
||
try:
|
||
raw = lokaal.ss.client.service.getUserDetailsByUsername(Config.SS_API_PW, kandidaat)
|
||
except Exception:
|
||
continue
|
||
if isinstance(raw, str) and raw.strip().startswith('{'):
|
||
gevonden = kandidaat
|
||
break
|
||
return sleutel, gevonden
|
||
|
||
with ThreadPoolExecutor(max_workers=4) as pool:
|
||
for sleutel, gevonden in pool.map(zoek, te_doen):
|
||
cache[sleutel] = (time.time(), gevonden)
|
||
uit[sleutel[0]] = gevonden
|
||
return uit
|