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)}"