277 lines
13 KiB
Python
277 lines
13 KiB
Python
"""
|
||
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
|