174 lines
7.9 KiB
Python
174 lines
7.9 KiB
Python
"""
|
||
Verzendlijsten (Google Groepen) van de huidige school beheren via de Admin SDK.
|
||
|
||
Directory API (scope admin.directory.group) → groepen en leden: bekijken, aanmaken, aanpassen
|
||
Groups Settings API (scope apps.groups.settings) → "wie mag e-mails sturen naar de lijst" (optioneel: ontbreekt de
|
||
scope, dan werkt de rest gewoon en wordt dit veld verborgen)
|
||
Beide scopes moeten in Google Admin › Beveiliging › API-beheer › Domeinbrede delegatie bij het serviceaccount staan.
|
||
"""
|
||
import re
|
||
import threading
|
||
|
||
from googleapiclient.discovery import build
|
||
|
||
from config import Config
|
||
from tenancy import google_credentials, school_singleton
|
||
|
||
GROUP_SCOPE = 'https://www.googleapis.com/auth/admin.directory.group'
|
||
SETTINGS_SCOPE = 'https://www.googleapis.com/auth/apps.groups.settings'
|
||
|
||
ROLLEN = [('MEMBER', 'Lid'), ('MANAGER', 'Beheerder'), ('OWNER', 'Eigenaar')]
|
||
WIE_MAILT = [
|
||
('ANYONE_CAN_POST', 'Iedereen, ook van buiten de school'),
|
||
('ALL_IN_DOMAIN_CAN_POST', 'Iedereen van de school'),
|
||
('ALL_MEMBERS_CAN_POST', 'Enkel leden van de lijst'),
|
||
('ALL_MANAGERS_CAN_POST', 'Enkel beheerders en eigenaars'),
|
||
]
|
||
EMAIL_RE = re.compile(r'^[^@\s]+@[^@\s]+\.[^@\s]+$')
|
||
|
||
|
||
class VerzendlijstFout(Exception):
|
||
pass
|
||
|
||
|
||
def _uitleg(e):
|
||
"""Google-fout → begrijpelijke melding."""
|
||
t = str(e)
|
||
laag = t.lower()
|
||
if 'unauthorized_client' in laag or 'insufficient authentication scopes' in laag or 'access_denied' in laag:
|
||
return ("Google weigert de toegang tot de groepen. Voeg in Google Admin › Beveiliging › API-beheer › "
|
||
f"Domeinbrede delegatie bij het serviceaccount de scope {GROUP_SCOPE} toe "
|
||
f"(en voor 'wie mag mailen' ook {SETTINGS_SCOPE}). Een wijziging kan enkele minuten duren.")
|
||
if 'entity already exists' in laag or 'duplicate' in laag or 'member already exists' in laag:
|
||
return "Dat adres bestaat al (als groep, gebruiker of lid)."
|
||
if 'resource not found' in laag or 'notfound' in laag.replace(' ', ''):
|
||
return "Niet gevonden in Google (bestaat het adres of de groep wel?)."
|
||
if 'domain' in laag and ('not found' in laag or 'invalid' in laag):
|
||
return "Dat domein hoort niet bij jullie Google Workspace."
|
||
m = re.search(r'"message":\s*"([^"]+)"', t) or re.search(r'returned "([^"]+)"', t)
|
||
return f"Google: {m.group(1) if m else t[:300]}"
|
||
|
||
|
||
def adressen(tekst):
|
||
"""Komma-, puntkomma-, spatie- of regelgescheiden lijst → [geldige adressen], [ongeldige stukken]."""
|
||
goed, fout = [], []
|
||
for d in re.split(r'[\s,;]+', str(tekst or '')):
|
||
d = d.strip().strip('<>').lower()
|
||
if not d:
|
||
continue
|
||
(goed if EMAIL_RE.match(d) else fout).append(d)
|
||
return list(dict.fromkeys(goed)), fout
|
||
|
||
|
||
class VerzendlijstenService:
|
||
def __new__(cls):
|
||
def init(inst):
|
||
inst._clients = threading.local() # Google-clients zijn niet thread-safe: één per thread
|
||
return school_singleton(cls, init)
|
||
|
||
def _api(self, naam, scope, api, versie):
|
||
client = getattr(self._clients, naam, None)
|
||
if client is None:
|
||
creds = google_credentials([scope], subject=Config.GOOGLE_ADMIN_EMAIL)
|
||
client = build(api, versie, credentials=creds, static_discovery=False, cache_discovery=False)
|
||
setattr(self._clients, naam, client)
|
||
return client
|
||
|
||
def _dir(self):
|
||
return self._api('directory', GROUP_SCOPE, 'admin', 'directory_v1')
|
||
|
||
def _settings(self):
|
||
return self._api('settings', SETTINGS_SCOPE, 'groupssettings', 'v1')
|
||
|
||
def _uitvoeren(self, verzoek):
|
||
try:
|
||
return verzoek.execute()
|
||
except Exception as e:
|
||
raise VerzendlijstFout(_uitleg(e))
|
||
|
||
# ---------------- groepen ----------------
|
||
def alle(self):
|
||
uit, token = [], None
|
||
while True:
|
||
res = self._uitvoeren(self._dir().groups().list(customer='my_customer', maxResults=200, pageToken=token,
|
||
orderBy='email'))
|
||
for g in res.get('groups', []):
|
||
uit.append({'email': g.get('email', ''), 'naam': g.get('name', ''),
|
||
'beschrijving': g.get('description', ''),
|
||
'leden': int(g.get('directMembersCount') or 0), 'aliassen': g.get('aliases', [])})
|
||
token = res.get('nextPageToken')
|
||
if not token:
|
||
return uit
|
||
|
||
def groep(self, email):
|
||
g = self._uitvoeren(self._dir().groups().get(groupKey=email))
|
||
return {'email': g.get('email', ''), 'naam': g.get('name', ''), 'beschrijving': g.get('description', ''),
|
||
'leden': int(g.get('directMembersCount') or 0), 'aliassen': g.get('aliases', [])}
|
||
|
||
def aanmaken(self, email, naam, beschrijving=''):
|
||
email = (email or '').strip().lower()
|
||
if not EMAIL_RE.match(email):
|
||
raise VerzendlijstFout("Geef een geldig e-mailadres voor de verzendlijst.")
|
||
if not (naam or '').strip():
|
||
raise VerzendlijstFout("Geef de verzendlijst een naam.")
|
||
self._uitvoeren(self._dir().groups().insert(body={'email': email, 'name': naam.strip(),
|
||
'description': (beschrijving or '').strip()}))
|
||
return email
|
||
|
||
def bijwerken(self, email, naam, beschrijving):
|
||
if not (naam or '').strip():
|
||
raise VerzendlijstFout("De naam mag niet leeg zijn.")
|
||
self._uitvoeren(self._dir().groups().patch(groupKey=email, body={
|
||
'name': naam.strip(), 'description': (beschrijving or '').strip()}))
|
||
|
||
# ---------------- wie mag mailen (Groups Settings API) ----------------
|
||
def wie_mailt(self, email):
|
||
"""Huidige instelling, of None als de Groups Settings-scope (nog) ontbreekt."""
|
||
try:
|
||
s = self._settings().groups().get(groupUniqueId=email, alt='json').execute()
|
||
return s.get('whoCanPostMessage')
|
||
except Exception:
|
||
return None
|
||
|
||
def zet_wie_mailt(self, email, waarde):
|
||
if waarde not in dict(WIE_MAILT):
|
||
raise VerzendlijstFout("Onbekende keuze voor 'wie mag mailen'.")
|
||
body = {'whoCanPostMessage': waarde}
|
||
if waarde == 'ANYONE_CAN_POST':
|
||
body['allowExternalMembers'] = 'false' # extern mailen mag, extern lid worden niet automatisch
|
||
self._uitvoeren(self._settings().groups().patch(groupUniqueId=email, body=body))
|
||
|
||
# ---------------- leden ----------------
|
||
def leden(self, email):
|
||
uit, token = [], None
|
||
while True:
|
||
res = self._uitvoeren(self._dir().members().list(groupKey=email, maxResults=200, pageToken=token))
|
||
for m in res.get('members', []):
|
||
uit.append({'email': m.get('email', ''), 'rol': m.get('role', 'MEMBER'),
|
||
'soort': m.get('type', 'USER'), 'status': m.get('status', '')})
|
||
token = res.get('nextPageToken')
|
||
if not token:
|
||
volgorde = {'OWNER': 0, 'MANAGER': 1, 'MEMBER': 2}
|
||
return sorted(uit, key=lambda m: (volgorde.get(m['rol'], 3), m['email']))
|
||
|
||
def leden_toevoegen(self, email, lijst, rol='MEMBER'):
|
||
"""Geeft (toegevoegd, [(adres, fout)])."""
|
||
if rol not in dict(ROLLEN):
|
||
rol = 'MEMBER'
|
||
gelukt, fouten = 0, []
|
||
for adres in lijst:
|
||
try:
|
||
self._uitvoeren(self._dir().members().insert(groupKey=email, body={'email': adres, 'role': rol}))
|
||
gelukt += 1
|
||
except VerzendlijstFout as e:
|
||
fouten.append((adres, str(e)))
|
||
return gelukt, fouten
|
||
|
||
def lid_rol(self, email, lid, rol):
|
||
if rol not in dict(ROLLEN):
|
||
raise VerzendlijstFout("Onbekende rol.")
|
||
self._uitvoeren(self._dir().members().patch(groupKey=email, memberKey=lid, body={'role': rol}))
|
||
|
||
def lid_verwijderen(self, email, lid):
|
||
self._uitvoeren(self._dir().members().delete(groupKey=email, memberKey=lid))
|