PIVOT 1.1.0
This commit is contained in:
marijn committed
commit
8a53234eb6
161 files changed
+31109
No files matched your search
@@ -0,0 +1,391 @@
|
||||
from googleapiclient.discovery import build
|
||||
from config import Config
|
||||
from tenancy import google_credentials
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import smtplib
|
||||
from email.mime.text import MIMEText
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
|
||||
class GoogleAdminService:
|
||||
def __init__(self):
|
||||
#lijst met scopes
|
||||
self.conf = Config()
|
||||
self.server = Config.GOOGLE_SMTP_SERVER
|
||||
self.port = Config.GOOGLE_SMTP_PORT
|
||||
self.sender = Config.GOOGLE_SMTP_SENDER_adres
|
||||
# Serviceaccount van de huidige school, gedelegeerd naar de admin van die school
|
||||
self.delegated_creds = google_credentials(self.conf.GOOGLE_API_SCOPES, subject=self.conf.GOOGLE_ADMIN_EMAIL)
|
||||
self.service = build('admin', 'directory_v1', credentials=self.delegated_creds,static_discovery=False)
|
||||
|
||||
#functie die passwoord aanpast
|
||||
def change_password(self, email, new_password):
|
||||
"""Wijzigt het wachtwoord van een gebruiker in Google Workspace."""
|
||||
try:
|
||||
self.service.users().update(
|
||||
userKey=email,
|
||||
body={'password': new_password, 'changePasswordAtNextLogin': True}
|
||||
).execute()
|
||||
return True, "Wachtwoord succesvol gewijzigd."
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
#Functie die info ophaalt
|
||||
def get_user_info(self, email):
|
||||
"""Haalt info op over een specifieke gebruiker."""
|
||||
try:
|
||||
user = self.service.users().get(userKey=email).execute()
|
||||
return user
|
||||
except:
|
||||
return None
|
||||
#Functie die nieuwe gebruiker aanmaakt
|
||||
def create_user(self, first_name, last_name, password, org_unit_path):
|
||||
try:
|
||||
primary_email = f"{first_name.lower()}.{last_name.lower()}@{self.conf.SCHOOL_DOMAIN}"
|
||||
user_body = {
|
||||
'primaryEmail': primary_email,
|
||||
'name': {'givenName': first_name.capitalize(), 'familyName': last_name.capitalize()},
|
||||
'password': password,
|
||||
'orgUnitPath': org_unit_path,
|
||||
'changePasswordAtNextLogin': True
|
||||
}
|
||||
self.service.users().insert(body=user_body).execute()
|
||||
return True, primary_email
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
#functie die opzoek of genruiger reeds aawezig is
|
||||
def search_users(self, query_string):
|
||||
"""
|
||||
Zoekt gebruikers in Google Workspace op basis van een query.
|
||||
Bijv: query_string="name:Jan*" of "email:jan@school.be"
|
||||
"""
|
||||
try:
|
||||
results = self.service.users().list(
|
||||
customer='my_customer',
|
||||
query=query_string,
|
||||
maxResults=10,
|
||||
orderBy='email'
|
||||
).execute()
|
||||
|
||||
return results.get('users', [])
|
||||
except Exception as e:
|
||||
print(f"Fout bij zoeken in Google: {e}")
|
||||
return []
|
||||
|
||||
#functie die toestelstatus aanpast (disabled/enabled)
|
||||
def toggle_device_status(self, device_id, lock=True):
|
||||
print("My debug message", flush=True)
|
||||
"""Blokkeert (disable) of deblokkeert (reenable) een ChromeOS device."""
|
||||
try:
|
||||
import sys
|
||||
#print die gebruikt word als debug check
|
||||
sys.stderr.write(f"DEBUG: Calling Google API for ID: {device_id}\n")
|
||||
action_type = 'disable' if lock else 'reenable'
|
||||
|
||||
|
||||
|
||||
self.service.chromeosdevices().action(
|
||||
customerId='my_customer',
|
||||
resourceId=device_id,
|
||||
body={'action': action_type}
|
||||
).execute()
|
||||
|
||||
return True, f"Toestel status succesvol gewijzigd naar {action_type}"
|
||||
except Exception as e:
|
||||
import sys
|
||||
sys.stderr.write(f"\n--- GOOGLE API ERROR: {str(e)} ---\n")
|
||||
return False, f"Google API Fout: {str(e)}"
|
||||
|
||||
#Functie die na het ingeven van de serienummer het DeviceID ophaalt (dit is een andere string dan het serienummer en essentieel om het toestel te kunnen blokkeren)
|
||||
def lijst_chromebooks(self, ous=None):
|
||||
"""Alle ingeschreven Chromebooks (niet uitgeschreven), eventueel beperkt tot OU's (incl. sub-OU's).
|
||||
Geeft een lijst dicts met serialNumber, model, status, orgUnitPath, annotatedAssetId."""
|
||||
velden = 'nextPageToken,chromeosdevices(serialNumber,model,status,orgUnitPath,annotatedAssetId)'
|
||||
toestellen, gezien = [], set()
|
||||
for ou in (ous or [None]):
|
||||
token = None
|
||||
while True:
|
||||
params = dict(customerId='my_customer', maxResults=200, projection='BASIC', fields=velden,
|
||||
pageToken=token)
|
||||
if ou:
|
||||
params.update(orgUnitPath=ou, includeChildOrgunits=True)
|
||||
resp = self.service.chromeosdevices().list(**params).execute()
|
||||
for d in resp.get('chromeosdevices', []):
|
||||
serie = (d.get('serialNumber') or '').strip()
|
||||
if not serie or serie in gezien or d.get('status') == 'DEPROVISIONED':
|
||||
continue
|
||||
gezien.add(serie)
|
||||
toestellen.append(d)
|
||||
token = resp.get('nextPageToken')
|
||||
if not token:
|
||||
break
|
||||
return toestellen
|
||||
|
||||
def chromebook_info(self, serienummer):
|
||||
"""Gebruiksinfo van één Chromebook: recente gebruikers, laatste synchronisatie, actieve tijd per dag.
|
||||
Geeft (dict, None) of (None, foutmelding)."""
|
||||
serie = str(serienummer or '').strip()
|
||||
if not serie:
|
||||
return None, "Dit toestel heeft geen serienummer."
|
||||
velden = ('chromeosdevices(serialNumber,model,status,orgUnitPath,lastSync,lastEnrollmentTime,osVersion,'
|
||||
'recentUsers,activeTimeRanges,annotatedUser,annotatedLocation)')
|
||||
devices = []
|
||||
for query in (f"id:{serie}", serie): # id: = zoeken op serienummer
|
||||
resp = self.service.chromeosdevices().list(customerId='my_customer', query=query, projection='FULL',
|
||||
maxResults=5, fields=velden).execute()
|
||||
devices = [d for d in resp.get('chromeosdevices', [])
|
||||
if (d.get('serialNumber') or '').strip().upper() == serie.upper()]
|
||||
if devices:
|
||||
break
|
||||
if not devices:
|
||||
return None, f"Geen Chromebook met serienummer {serie} gevonden in Google Admin."
|
||||
d = devices[0]
|
||||
return {
|
||||
'serienummer': d.get('serialNumber'),
|
||||
'model': d.get('model'),
|
||||
'status': d.get('status'),
|
||||
'ou': d.get('orgUnitPath'),
|
||||
'os': d.get('osVersion'),
|
||||
'laatste_sync': d.get('lastSync'),
|
||||
'ingeschreven': d.get('lastEnrollmentTime'),
|
||||
'gebruikers': [{'email': u.get('email') or '', 'beheerd': u.get('type') == 'USER_TYPE_MANAGED'}
|
||||
for u in d.get('recentUsers', [])],
|
||||
'actief': sorted(({'datum': r.get('date'), 'minuten': round(int(r.get('activeTime') or 0) / 60000)}
|
||||
for r in d.get('activeTimeRanges', [])), key=lambda r: r['datum'] or '',
|
||||
reverse=True)[:14],
|
||||
}, None
|
||||
|
||||
def zoek_chromebook(self, serienummer):
|
||||
"""(deviceId, status) van de Chromebook met exact dit serienummer, of (None, foutmelding).
|
||||
Status: ACTIVE, DISABLED, DEPROVISIONED, ..."""
|
||||
serie = str(serienummer or '').strip().replace('\t', '').replace('\n', '').replace('\r', '')
|
||||
if not serie:
|
||||
return None, "Dit toestel heeft geen serienummer."
|
||||
gevonden = []
|
||||
for query in (f"id:{serie}", serie): # id: = zoeken op serienummer
|
||||
resp = self.service.chromeosdevices().list(customerId='my_customer', query=query, maxResults=10,
|
||||
projection='BASIC',
|
||||
fields='chromeosdevices(deviceId,serialNumber,status)').execute()
|
||||
gevonden = [d for d in resp.get('chromeosdevices', [])
|
||||
if (d.get('serialNumber') or '').strip().upper() == serie.upper()]
|
||||
if gevonden:
|
||||
break
|
||||
if not gevonden:
|
||||
return None, f"Geen Chromebook met serienummer {serie} gevonden in Google Admin."
|
||||
# Bij een opnieuw ingeschreven toestel: de niet-uitgeschreven versie
|
||||
d = next((x for x in gevonden if x.get('status') != 'DEPROVISIONED'), gevonden[0])
|
||||
return {'id': d.get('deviceId'), 'status': d.get('status')}, None
|
||||
|
||||
def get_device_id_by_serial(self, serienummer):
|
||||
"""Zoekt het deviceId op basis van het fysieke serienummer."""
|
||||
print(f"DEBUG: Zoeken naar serienummer: '{serienummer}'", flush=True)
|
||||
try:
|
||||
|
||||
# Verwijder spaties, tabs (\t) en newlines (\n)
|
||||
clean_serial = str(serienummer).strip().replace('\t', '').replace('\n', '').replace('\r', '')
|
||||
|
||||
# Voer de API call uit met het schone serienummer
|
||||
results = self.service.chromeosdevices().list(
|
||||
customerId='my_customer',
|
||||
query=f"{clean_serial}"
|
||||
).execute()
|
||||
|
||||
devices = results.get('chromeosdevices', [])
|
||||
|
||||
if devices:
|
||||
dev_id = devices[0].get('deviceId')
|
||||
sys.stderr.write(f"DEBUG: Toestel gevonden! ID: {dev_id}\n")
|
||||
return dev_id, None
|
||||
|
||||
return None, f"Device met serienummer {clean_serial} niet gevonden."
|
||||
|
||||
except Exception as e:
|
||||
sys.stderr.write(f"DEBUG: Fout in get_device_id_by_serial: {str(e)}\n")
|
||||
return None, str(e)
|
||||
|
||||
def list_org_units(self):
|
||||
"""Haalt alle Organizational Units op en sorteert ze alfabetisch."""
|
||||
try:
|
||||
results = self.service.orgunits().list(customerId='my_customer', type='all').execute()
|
||||
org_units = results.get('organizationUnits', [])
|
||||
|
||||
# Sorteren op het volledige pad (bijv. /Leerlingen komt na /Beheer)
|
||||
org_units.sort(key=lambda x: x.get('orgUnitPath', '').lower())
|
||||
|
||||
return org_units
|
||||
except Exception as e:
|
||||
import sys
|
||||
sys.stderr.write(f"Fout bij ophalen OUs: {e}\n")
|
||||
return []
|
||||
|
||||
def move_users_ou(self, source_ou, target_ou, only_enrolled=True, enrolled_emails=None):
|
||||
"""
|
||||
Verplaatst gebruikers van source naar target.
|
||||
enrolled_emails: een lijst met e-mailadressen die volgens Informat/database ingeschreven zijn.
|
||||
"""
|
||||
users_moved = 0
|
||||
try:
|
||||
# We zoeken specifiek gebruikers in de bron-OU om API-quota te besparen
|
||||
query = f"orgUnitPath='{source_ou}'"
|
||||
results = self.service.users().list(customer='my_customer', query=query).execute()
|
||||
users = results.get('users', [])
|
||||
|
||||
for user in users:
|
||||
user_email = user.get('primaryEmail')
|
||||
|
||||
# Logica voor de radiobuttons uit je HTML
|
||||
should_move = False
|
||||
if not only_enrolled:
|
||||
should_move = True # "Alle leerlingen"
|
||||
elif enrolled_emails and user_email in enrolled_emails:
|
||||
should_move = True # "Enkel ingeschreven leerlingen"
|
||||
|
||||
if should_move:
|
||||
self.service.users().update(
|
||||
userKey=user_email,
|
||||
body={'orgUnitPath': target_ou}
|
||||
).execute()
|
||||
users_moved += 1
|
||||
|
||||
return True, f"{users_moved} gebruikers succesvol verplaatst."
|
||||
except Exception as e:
|
||||
return False, f"Fout bij verplaatsen: {str(e)}"
|
||||
|
||||
def move_chromebooks_ou(self, source_ou, target_ou):
|
||||
"""Verplaatst alle Chromebooks van de bron-OU naar de doel-OU."""
|
||||
devices_moved = 0
|
||||
next_page_token = None
|
||||
try:
|
||||
while True:
|
||||
# We halen alle toestellen op en filteren handmatig op de OU
|
||||
results = self.service.chromeosdevices().list(
|
||||
customerId='my_customer',
|
||||
pageToken=next_page_token,
|
||||
maxResults=100
|
||||
).execute()
|
||||
|
||||
devices = results.get('chromeosdevices', [])
|
||||
for device in devices:
|
||||
if device.get('orgUnitPath') == source_ou:
|
||||
device_id = device['deviceId']
|
||||
self.service.chromeosdevices().update(
|
||||
customerId='my_customer',
|
||||
deviceId=device_id,
|
||||
body={'orgUnitPath': target_ou}
|
||||
).execute()
|
||||
devices_moved += 1
|
||||
|
||||
next_page_token = results.get('nextPageToken')
|
||||
if not next_page_token:
|
||||
break
|
||||
return True, f"{devices_moved} Chromebooks succesvol verplaatst."
|
||||
except Exception as e:
|
||||
return False, f"Fout bij verplaatsen Chromebooks: {str(e)}"
|
||||
|
||||
def toggle_ou_devices_status(self, org_unit_path, block=True):
|
||||
"""
|
||||
Blokkeert of deblokkeert alle ChromeOS toestellen in een specifieke OU.
|
||||
Voorkomt 412 errors door de huidige status te controleren.
|
||||
"""
|
||||
devices_processed = 0
|
||||
action_type = 'disable' if block else 'reenable'
|
||||
# De status die het toestel AL heeft als de actie overbodig is
|
||||
forbidden_status = 'DISABLED' if block else 'ACTIVE'
|
||||
|
||||
next_page_token = None
|
||||
|
||||
try:
|
||||
while True:
|
||||
results = self.service.chromeosdevices().list(
|
||||
customerId='my_customer',
|
||||
pageToken=next_page_token,
|
||||
maxResults=100
|
||||
).execute()
|
||||
|
||||
devices = results.get('chromeosdevices', [])
|
||||
|
||||
for device in devices:
|
||||
if device.get('orgUnitPath') == org_unit_path:
|
||||
current_status = device.get('status') # bijv. 'ACTIVE' of 'DISABLED'
|
||||
device_id = device['deviceId']
|
||||
serial = device.get('serialNumber', 'Onbekend')
|
||||
|
||||
# Alleen actie ondernemen als de status nog niet is
|
||||
if current_status != forbidden_status:
|
||||
try:
|
||||
self.service.chromeosdevices().action(
|
||||
customerId='my_customer',
|
||||
resourceId=device_id,
|
||||
body={'action': action_type}
|
||||
).execute()
|
||||
devices_processed += 1
|
||||
except Exception as e:
|
||||
print(f"Kon {serial} niet wijzigen: {e}")
|
||||
else:
|
||||
print(f"Overgeslagen: {serial} is al {current_status}")
|
||||
|
||||
next_page_token = results.get('nextPageToken')
|
||||
if not next_page_token:
|
||||
break
|
||||
|
||||
status_tekst = "geblokkeerd" if block else "gedeblokkeerd"
|
||||
return True, f"{devices_processed} toestellen succesvol {status_tekst}. Toestellen die al de juiste status hadden zijn overgeslagen."
|
||||
except Exception as e:
|
||||
return False, f"Algemene fout: {str(e)}"
|
||||
|
||||
def send_custom_mail(self, receiver_email, subject, body):
|
||||
msg = MIMEMultipart()
|
||||
msg['From'] = self.sender
|
||||
msg['To'] = receiver_email
|
||||
msg['Subject'] = subject
|
||||
|
||||
msg.attach(MIMEText(body, 'html')) # We sturen het als HTML voor mooie opmaak
|
||||
|
||||
try:
|
||||
with smtplib.SMTP(self.server, self.port) as server:
|
||||
server.starttls()
|
||||
# Geen login nodig voor de relay indien geconfigureerd op IP
|
||||
wachtwoord = Config.GOOGLE_SMTP_EMAIL_PASSWORD
|
||||
if wachtwoord and wachtwoord != 'None':
|
||||
server.login(self.sender, wachtwoord)
|
||||
server.sendmail(self.sender, receiver_email, msg.as_string())
|
||||
return True, "Email succesvol verzonden"
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
def archive_inactive_users(self, source_ou, days_inactive):
|
||||
"""Archiveert gebruikers die X dagen niet zijn ingelogd, zonder van OU te wisselen."""
|
||||
users_archived = 0
|
||||
cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_inactive)
|
||||
|
||||
try:
|
||||
# Zoek gebruikers in de bron-OU
|
||||
query = f"orgUnitPath='{source_ou}'"
|
||||
results = self.service.users().list(customer='my_customer', query=query).execute()
|
||||
users = results.get('users', [])
|
||||
|
||||
for user in users:
|
||||
last_login_str = user.get('lastLoginTime')
|
||||
user_email = user.get('primaryEmail')
|
||||
|
||||
should_archive = False
|
||||
if last_login_str:
|
||||
last_login_dt = datetime.fromisoformat(last_login_str.replace('Z', '+00:00'))
|
||||
if last_login_dt < cutoff_date:
|
||||
should_archive = True
|
||||
else:
|
||||
# Nooit ingelogd = inactief
|
||||
should_archive = True
|
||||
|
||||
if should_archive:
|
||||
# We sturen enkel 'archived': True en GEEN 'orgUnitPath'
|
||||
self.service.users().update(
|
||||
userKey=user_email,
|
||||
body={'archived': True}
|
||||
).execute()
|
||||
users_archived += 1
|
||||
|
||||
return True, f"{users_archived} gebruikers succesvol gearchiveerd (OU ongewijzigd)."
|
||||
except Exception as e:
|
||||
return False, f"Fout bij archiveren: {str(e)}"
|
||||
|
||||
Reference in new issue
Block a user