391 lines
18 KiB
Python
391 lines
18 KiB
Python
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)}"
|
|
|