""" Google Workspace for Education – licentie-overzicht. Haalt via de Enterprise License Manager API op welke gebruikers een licentie hebben. Google geeft het aantal AANGEKOCHTE licenties niet vrij via de API; dat aantal wordt in PIVOT bijgehouden (tabel google_licenties) en door een admin ingevuld. Vereist (eenmalig, Google Admin-console): Beveiliging > Toegangs- en gegevensbeheer > API-beheer > Domeinbrede delegatie > service-account bewerken en deze scope toevoegen: https://www.googleapis.com/auth/apps.licensing """ import sys import time from googleapiclient.discovery import build from config import Config from database import execute_query from tenancy import google_credentials LICENSING_SCOPE = 'https://www.googleapis.com/auth/apps.licensing' # Product- en SKU-ID's volgens Google: # https://developers.google.com/workspace/admin/licensing/v1/how-tos/products EDU_PRODUCT_ID = '101031' SKU_NAMES = { '1010310008': 'Education Plus', '1010310009': 'Education Plus (Staff)', '1010310010': 'Education Plus (Extra Student)', '1010310005': 'Education Standard', '1010310006': 'Education Standard (Staff)', '1010310007': 'Education Standard (Extra Student)', } # Deze SKU's worden altijd getoond, ook als er (nog) niemand een licentie heeft DEFAULT_SKUS = ['1010310008'] # Licenties die de scholengroep niet gebruikt: niet tonen, tenzij iemand er toch (per vergissing) één heeft VERBORGEN_SKUS = {'1010310009'} # Education Plus (Staff) CACHE_SECONDS = 10 * 60 def _soort_fout(e): """'scope' = delegatie/scope ontbreekt, 'api' = API niet ingeschakeld, anders None.""" laag = str(e).lower() if 'unauthorized_client' in laag or 'access_denied' in laag or 'invalid_scope' in laag: return 'scope' if 'accessnotconfigured' in laag or 'has not been used' in laag or 'is disabled' in laag \ or 'service_disabled' in laag: return 'api' return None def _google_reden(e): """De foutboodschap van Google zelf (zonder de lange URL).""" try: import json inhoud = json.loads(getattr(e, 'content', b'') or b'{}') fout = inhoud.get('error') or {} if isinstance(fout, dict) and fout.get('message'): return f"{getattr(getattr(e, 'resp', None), 'status', '')} {fout['message']}".strip() except Exception: pass return str(e)[:250] class GoogleLicenseService: def __init__(self, directory_service=None): self._cache = None # (timestamp, data) self._licensing = None self._klant = None # customerId die werkt # Bestaande Directory-client hergebruiken (scopes zijn daar al geautoriseerd) self._directory = directory_service # ---------- API-clients (lui aangemaakt, zodat de app ook start zonder scope) ---------- def _creds(self, scope): return google_credentials([scope], subject=Config.GOOGLE_ADMIN_EMAIL) def _licensing_api(self): if self._licensing is None: self._licensing = build('licensing', 'v1', credentials=self._creds(LICENSING_SCOPE), static_discovery=False, cache_discovery=False) return self._licensing def _directory_api(self): if self._directory is None: from services.google_service import GoogleAdminService self._directory = GoogleAdminService().service return self._directory # ---------- Database: aangekochte aantallen ---------- @staticmethod def ensure_table(): execute_query(""" CREATE TABLE IF NOT EXISTS google_licenties ( SKU_ID VARCHAR(32) NOT NULL PRIMARY KEY, Naam VARCHAR(128) NOT NULL, Aangekocht INT NOT NULL DEFAULT 0, Bijgewerkt DATETIME NULL, Bijgewerkt_door VARCHAR(100) NULL )""", commit=True) def get_purchased(self): rows = execute_query("SELECT SKU_ID, Naam, Aangekocht, Bijgewerkt, Bijgewerkt_door FROM google_licenties") return {r['SKU_ID']: r for r in rows} def set_purchased(self, sku_id, aantal, gebruiker): naam = SKU_NAMES.get(sku_id, sku_id) execute_query(""" INSERT INTO google_licenties (SKU_ID, Naam, Aangekocht, Bijgewerkt, Bijgewerkt_door) VALUES (%s, %s, %s, NOW(), %s) ON DUPLICATE KEY UPDATE Aangekocht = VALUES(Aangekocht), Naam = VALUES(Naam), Bijgewerkt = NOW(), Bijgewerkt_door = VALUES(Bijgewerkt_door) """, (sku_id, naam, int(aantal), gebruiker), commit=True) # ---------- Google: toegewezen licenties ---------- def _klant_kandidaten(self): """Waarden om als customerId te proberen. De Licensing API wil het PRIMAIRE domein van de Google-klant (of de klant-ID, bv. C01abc23d). Het domein dat in PIVOT bij de school staat, is soms een secundair domein of alias: dan antwoordt Google met 403/400. Daarom ook de klant-ID van de beheerder proberen.""" kandidaten = [Config.SCHOOL_DOMAIN] try: u = self._directory_api().users().get(userKey=Config.GOOGLE_ADMIN_EMAIL, fields='customerId,primaryEmail').execute() kandidaten += [u.get('customerId'), (u.get('primaryEmail') or '').partition('@')[2]] except Exception as e: sys.stderr.write(f"Licenties: klant-ID van de beheerder niet op te halen: {e}\n") kandidaten.append((Config.GOOGLE_ADMIN_EMAIL or '').partition('@')[2]) return [k for k in dict.fromkeys(k.strip() for k in kandidaten if k and k.strip())] def _list_assignments(self): api = self._licensing_api() laatste = None for klant in (self._klant_kandidaten() if not getattr(self, '_klant', None) else [self._klant]): assignments, token = [], None try: while True: resp = api.licenseAssignments().listForProduct( productId=EDU_PRODUCT_ID, customerId=klant, maxResults=1000, pageToken=token).execute() assignments.extend(resp.get('items', [])) token = resp.get('nextPageToken') if not token: self._klant = klant # onthouden welke waarde werkt return assignments except Exception as e: status = getattr(getattr(e, 'resp', None), 'status', None) if status not in (400, 403, 404) or _soort_fout(e) in ('scope', 'api'): raise sys.stderr.write(f"Licenties met customerId={klant} mislukt: {e}\n") laatste = e raise laatste or RuntimeError("Geen Google-klant gevonden om licenties voor op te vragen.") def _uitleg(self, e): """Begrijpelijke melding met de echte reden van Google en wat je kan nakijken.""" from tenancy import huidige_school sleutel = huidige_school().google_sleutel or {} client_id = sleutel.get('client_id', '?') project = sleutel.get('project_id', '?') admin = Config.GOOGLE_ADMIN_EMAIL soort = _soort_fout(e) reden = _google_reden(e) if soort == 'scope': return (f"Google weigert de toegang (unauthorized_client): voor client-ID {client_id} staat de scope " f"{LICENSING_SCOPE} niet in de domeinbrede delegatie. Kijk na in de Admin-console van het domein van " f"{admin} (Beveiliging › API-beheer › Domeinbrede delegatie) dat de scope bij exact deze client-ID " "staat — elke school met een eigen JSON-sleutel heeft een eigen client-ID. Een wijziging kan tot " "een uur duren.") if soort == 'api': return (f"De Enterprise License Manager API staat niet aan in Google Cloud-project '{project}' " "(console.cloud.google.com › API's en services › Bibliotheek › 'Enterprise License Manager API' " f"› Inschakelen). Google: {reden}") return (f"Google weigert de licenties ({reden}). De delegatie zelf werkt, dus de scope is in orde. " f"Mogelijke oorzaken: {admin} is geen hoofdbeheerder (super admin) van het domein, of het domein " f"'{Config.SCHOOL_DOMAIN}' is niet het primaire domein van de Google-klant (ook de klant-ID werd " f"geprobeerd), of deze school heeft geen Education Standard/Plus-licenties. Service-account: " f"client-ID {client_id}, project '{project}'.") def _user_details(self): """Alle gebruikers met OU en status, zodat we per licentie context kunnen tonen.""" api = self._directory_api() users, token = {}, None while True: resp = api.users().list( customer='my_customer', maxResults=500, pageToken=token, projection='basic', fields='nextPageToken,users(primaryEmail,name/fullName,orgUnitPath,suspended,archived,lastLoginTime)' ).execute() for u in resp.get('users', []): users[u['primaryEmail'].lower()] = u token = resp.get('nextPageToken') if not token: return users def get_overview(self, force=False): """ Geeft (data, fout) terug. data = { 'skus': [ {sku_id, naam, in_gebruik, aangekocht, vrij, procent, geschorst, ...}, ... ], 'users': [ {email, naam, ou, sku_id, sku_naam, geschorst, gearchiveerd, laatste_login}, ... ], 'opgehaald': timestamp } """ if not force and self._cache and time.time() - self._cache[0] < CACHE_SECONDS: return self._cache[1], None try: assignments = self._list_assignments() except Exception as e: sys.stderr.write(f"Licenties ophalen mislukt: {e}\n") self._licensing = None # volgende keer opnieuw opbouwen (bv. na een nieuwe JSON-sleutel) self._klant = None try: return None, self._uitleg(e) except Exception: return None, str(e)[:400] try: directory = self._user_details() except Exception as e: sys.stderr.write(f"Gebruikersdetails ophalen mislukt (niet kritisch): {e}\n") directory = {} try: purchased = self.get_purchased() except Exception as e: sys.stderr.write(f"Tabel google_licenties niet leesbaar: {e}\n") purchased = {} users, counts, suspended_counts = [], {}, {} for a in assignments: sku = a.get('skuId') email = (a.get('userId') or '').lower() d = directory.get(email, {}) geschorst = bool(d.get('suspended')) gearchiveerd = bool(d.get('archived')) counts[sku] = counts.get(sku, 0) + 1 if geschorst or gearchiveerd: suspended_counts[sku] = suspended_counts.get(sku, 0) + 1 users.append({ 'email': email, 'naam': (d.get('name') or {}).get('fullName', ''), 'ou': d.get('orgUnitPath', ''), 'sku_id': sku, 'sku_naam': SKU_NAMES.get(sku, a.get('skuName', sku)), 'geschorst': geschorst, 'gearchiveerd': gearchiveerd, 'onbekend': not d, 'laatste_login': (d.get('lastLoginTime') or '')[:10].replace('1970-01-01', ''), }) users.sort(key=lambda u: (u['sku_naam'], u['ou'], u['email'])) sku_ids = list(dict.fromkeys(DEFAULT_SKUS + list(counts) + list(purchased))) skus = [] for sku in sku_ids: in_gebruik = counts.get(sku, 0) if sku in VERBORGEN_SKUS and not in_gebruik: continue aangekocht = int(purchased.get(sku, {}).get('Aangekocht') or 0) vrij = aangekocht - in_gebruik procent = round(in_gebruik / aangekocht * 100) if aangekocht else None skus.append({ 'sku_id': sku, 'naam': SKU_NAMES.get(sku, purchased.get(sku, {}).get('Naam', sku)), 'in_gebruik': in_gebruik, 'aangekocht': aangekocht, 'vrij': vrij, 'procent': procent, 'geschorst': suspended_counts.get(sku, 0), 'bijgewerkt': purchased.get(sku, {}).get('Bijgewerkt'), 'bijgewerkt_door': purchased.get(sku, {}).get('Bijgewerkt_door'), }) data = {'skus': skus, 'users': users, 'opgehaald': time.time()} self._cache = (time.time(), data) return data, None def clear_cache(self): self._cache = None