Files
Pivot/services/handtekening_service.py
2026-10-14 13:34:37 +02:00

444 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Gmail › Handtekeningen: één HTML-sjabloon per school, ingevuld per gebruiker en ingesteld in Gmail.
- Gebruikers: alle accounts in de gekozen OU's (Directory API, optioneel met de onderliggende OU's).
- Functie en telefoon komen uit Google (gebruikersprofiel: organizations[].title en phones[]) en worden daar ook
bewaard als je ze in PIVOT aanpast. Zo staan ze ook in het adresboek van de organisatie.
- Toepassen: Gmail API users.settings.sendAs.patch(signature) namens de gebruiker, voor het hoofdadres en de aliassen.
- Logo: geüpload in PIVOT (publiek op /handtekening-logo/<school>/<sleutel>) of een eigen publieke link.
Gmail toont geen ingesloten (base64) afbeeldingen in handtekeningen, enkel afbeeldingen via een link.
- Nachtelijk opnieuw toepassen (schakelaar): gebruikers kunnen hun handtekening in Gmail zelf wijzigen,
zo staat de juiste er de volgende ochtend terug.
Vereist (domeinbrede delegatie): https://www.googleapis.com/auth/gmail.settings.basic
en voor functie/telefoon https://www.googleapis.com/auth/admin.directory.user (staat al in de standaardlijst).
"""
import hashlib
import html
import json
import re
import secrets
import sys
import threading
import time
from datetime import datetime
from config import Config
from database import execute_query
from tenancy import school_singleton, met_school, google_credentials, huidige_school
SCOPE = 'https://www.googleapis.com/auth/gmail.settings.basic'
MAX_TEKENS = 10000 # limiet van Gmail voor een handtekening
MAX_LOGO = 512 * 1024
LOGO_TYPES = {'image/png': 'png', 'image/jpeg': 'jpg', 'image/gif': 'gif'}
NACHT_UUR_VAN, NACHT_UUR_TOT = 3, 6
VELDEN = [ # (sleutel, uitleg)
('voornaam', 'Voornaam'), ('achternaam', 'Achternaam'), ('naam', 'Voornaam en achternaam'),
('functie', 'Functie (uit Google)'), ('telefoon', 'Telefoon (uit Google)'), ('email', 'E-mailadres'),
('schoolnaam', 'Naam van de school'), ('schooladres', 'Adres van de school'), ('logo', 'Logo (afbeelding)'),
]
STANDAARD_SJABLOON = """<table cellpadding="0" cellspacing="0" style="font-family:Arial,Helvetica,sans-serif;font-size:13px;color:#1f2937">
<tr>
<td style="padding-right:14px;vertical-align:top">{logo}</td>
<td style="border-left:2px solid #1d6fa5;padding-left:14px;vertical-align:top">
<div style="font-size:15px;font-weight:bold;color:#1d6fa5">{voornaam} {achternaam}</div>
{#functie}<div style="color:#4b5563">{functie}</div>{/functie}
<div style="margin-top:6px;font-weight:bold">{schoolnaam}</div>
<div>{schooladres}</div>
{#telefoon}<div>Tel. {telefoon}</div>{/telefoon}
<div><a href="mailto:{email}" style="color:#1d6fa5;text-decoration:none">{email}</a></div>
</td>
</tr>
</table>"""
class HandtekeningFout(Exception):
pass
# ---------------------------------------------------------------------
# Opslag (database van de school)
# ---------------------------------------------------------------------
def ensure_tables():
execute_query("""
CREATE TABLE IF NOT EXISTS gmail_handtekening (
ID TINYINT NOT NULL PRIMARY KEY,
Sjabloon MEDIUMTEXT NULL,
OUs TEXT NULL,
Met_sub_ous TINYINT NOT NULL DEFAULT 1,
Logo MEDIUMBLOB NULL,
Logo_type VARCHAR(40) NULL,
Logo_sleutel VARCHAR(40) NULL,
Logo_url VARCHAR(500) NULL,
Logo_breedte INT NOT NULL DEFAULT 120,
Publiek_adres VARCHAR(255) NULL,
Nachtelijk TINYINT NOT NULL DEFAULT 0,
Laatste_nacht DATE NULL,
Nacht_resultaat VARCHAR(500) NULL,
Bijgewerkt DATETIME NULL,
Door VARCHAR(255) NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""", commit=True)
execute_query("""
CREATE TABLE IF NOT EXISTS gmail_handtekening_log (
Email VARCHAR(255) NOT NULL PRIMARY KEY,
Toegepast DATETIME NOT NULL,
Ok TINYINT NOT NULL,
Melding VARCHAR(500) NULL,
Hash CHAR(40) NULL,
Door VARCHAR(255) NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""", commit=True)
execute_query("INSERT IGNORE INTO gmail_handtekening (ID, Sjabloon) VALUES (1, %s)", (STANDAARD_SJABLOON,), commit=True)
def instellingen():
ensure_tables()
r = (execute_query("SELECT ID, Sjabloon, OUs, Met_sub_ous, Logo_type, Logo_sleutel, Logo_url, Logo_breedte, "
"Publiek_adres, Nachtelijk, Laatste_nacht, Nacht_resultaat, Bijgewerkt, Door, "
"Logo IS NOT NULL AS heeft_logo FROM gmail_handtekening WHERE ID = 1") or [{}])[0]
try:
r['ous'] = [o for o in json.loads(r.get('OUs') or '[]') if isinstance(o, str)]
except ValueError:
r['ous'] = []
r['logo_link'] = logo_link(r)
return r
def bewaar(sjabloon, ous, met_sub, logo_url, logo_breedte, publiek_adres, door):
sjabloon = (sjabloon or '').strip()
if len(sjabloon) > 50000:
raise HandtekeningFout("Het sjabloon is te lang.")
execute_query("UPDATE gmail_handtekening SET Sjabloon = %s, OUs = %s, Met_sub_ous = %s, Logo_url = %s, "
"Logo_breedte = %s, Publiek_adres = %s, Bijgewerkt = NOW(), Door = %s WHERE ID = 1",
(sjabloon, json.dumps(sorted(set(ous))), 1 if met_sub else 0, (logo_url or '').strip() or None,
max(20, min(600, int(logo_breedte or 120))), (publiek_adres or '').strip().rstrip('/') or None, door),
commit=True)
def bewaar_logo(data, mimetype):
if mimetype not in LOGO_TYPES:
raise HandtekeningFout("Enkel PNG, JPG of GIF als logo.")
if len(data) > MAX_LOGO:
raise HandtekeningFout(f"Het logo is te groot (max. {MAX_LOGO // 1024} kB).")
execute_query("UPDATE gmail_handtekening SET Logo = %s, Logo_type = %s, Logo_sleutel = %s WHERE ID = 1",
(data, mimetype, secrets.token_hex(12)), commit=True)
def verwijder_logo():
execute_query("UPDATE gmail_handtekening SET Logo = NULL, Logo_type = NULL, Logo_sleutel = NULL WHERE ID = 1",
commit=True)
def logo_bestand(sleutel):
ensure_tables()
r = execute_query("SELECT Logo, Logo_type FROM gmail_handtekening WHERE ID = 1 AND Logo_sleutel = %s", (sleutel,))
return (r[0]['Logo'], r[0]['Logo_type']) if r and r[0]['Logo'] else (None, None)
def logo_link(inst):
"""Publieke link naar het logo: eigen link, of het geüploade logo via PIVOT."""
if inst.get('Logo_url'):
return inst['Logo_url']
if inst.get('heeft_logo') and inst.get('Logo_sleutel') and inst.get('Publiek_adres'):
ext = LOGO_TYPES.get(inst.get('Logo_type'), 'png')
return f"{inst['Publiek_adres']}/handtekening-logo/{huidige_school().code}/{inst['Logo_sleutel']}.{ext}"
return ''
def zet_nacht(aan, door):
execute_query("UPDATE gmail_handtekening SET Nachtelijk = %s, Door = %s WHERE ID = 1", (1 if aan else 0, door),
commit=True)
def logboek():
ensure_tables()
return {r['Email'].lower(): r for r in execute_query("SELECT * FROM gmail_handtekening_log") or []}
# ---------------------------------------------------------------------
# Sjabloon invullen
# ---------------------------------------------------------------------
def gegevens(gebruiker):
s = huidige_school()
return {'voornaam': gebruiker.get('voornaam', ''), 'achternaam': gebruiker.get('achternaam', ''),
'naam': f"{gebruiker.get('voornaam', '')} {gebruiker.get('achternaam', '')}".strip(),
'functie': gebruiker.get('functie', ''), 'telefoon': gebruiker.get('telefoon', ''),
'email': gebruiker.get('email', ''), 'schoolnaam': s.naam or '', 'schooladres': s.get('ADRES') or ''}
def vul_in(sjabloon, waarden, inst):
"""{veld} → waarde (HTML-veilig); {#veld}...{/veld} valt weg als het veld leeg is; {logo} → <img>."""
def blok(m):
return m.group(2) if (waarden.get(m.group(1)) or '').strip() or (m.group(1) == 'logo' and inst['logo_link']) else ''
uit = re.sub(r'\{#(\w+)\}(.*?)\{/\1\}', blok, sjabloon or '', flags=re.S)
link = inst.get('logo_link') or ''
logo = (f'<img src="{html.escape(link)}" width="{int(inst.get("Logo_breedte") or 120)}" alt="{html.escape(waarden.get("schoolnaam", ""))}" '
f'style="display:block;border:0;max-width:100%;height:auto">') if link else ''
def veld(m):
k = m.group(1)
if k == 'logo':
return logo
if k in waarden:
w = html.escape(waarden[k] or '')
return w.replace('\n', '<br>') if k == 'schooladres' else w
return m.group(0)
return re.sub(r'\{(\w+)\}', veld, uit)
# ---------------------------------------------------------------------
# Google: gebruikers, functie en telefoon
# ---------------------------------------------------------------------
def _directory():
from googleapiclient.discovery import build
creds = google_credentials(Config.GOOGLE_API_SCOPES, subject=Config.GOOGLE_ADMIN_EMAIL)
return build('admin', 'directory_v1', credentials=creds, static_discovery=False, cache_discovery=False)
def _gmail(mailbox):
from googleapiclient.discovery import build
creds = google_credentials([SCOPE], subject=mailbox)
try:
return build('gmail', 'v1', credentials=creds, static_discovery=True, cache_discovery=False)
except Exception:
return build('gmail', 'v1', credentials=creds, static_discovery=False, cache_discovery=False)
def vriendelijk(e):
t = str(e)
laag = t.lower()
if 'unauthorized_client' in laag or 'access_denied' in laag:
return (f"Google weigert de toegang: voeg in de Admin-console › Beveiliging › API-beheer › Domeinbrede delegatie "
f"het bereik {SCOPE} toe bij het serviceaccount van PIVOT.")
if 'has not been used' in laag or 'is disabled' in laag or 'accessnotconfigured' in laag:
return "De Gmail API staat niet aan in het Google Cloud-project van het serviceaccount."
if 'mail service not enabled' in laag or 'failedprecondition' in laag:
return "Deze gebruiker heeft geen Gmail (geen licentie of Gmail uitgeschakeld)."
if 'not authorized to access this resource' in laag or 'insufficient' in laag:
return "Google weigert het: de beheerder mag deze gebruiker niet aanpassen (scope admin.directory.user?)."
return t[:250]
def ou_lijst():
try:
resp = _directory().orgunits().list(customerId='my_customer', type='all').execute()
except Exception as e:
raise HandtekeningFout(f"OU's ophalen mislukt: {vriendelijk(e)}")
return sorted([o['orgUnitPath'] for o in resp.get('organizationUnits', [])] + ['/'], key=str.lower)
def _functie(u):
orgs = u.get('organizations') or []
org = next((o for o in orgs if o.get('primary')), orgs[0] if orgs else {})
return (org.get('title') or '').strip()
def _telefoon(u):
tel = u.get('phones') or []
t = next((p for p in tel if p.get('type') == 'work'), None) or next((p for p in tel if p.get('primary')), None) \
or (tel[0] if tel else {})
return (t.get('value') or '').strip()
def gebruikers(ous, met_sub=True):
"""Alle (niet-geschorste) gebruikers in de OU's: [{'email','voornaam','achternaam','ou','functie','telefoon'}]."""
if not ous:
return []
api, uit, gezien = _directory(), [], set()
for ou in ous:
token = None
while True:
try:
params = dict(customer='my_customer', maxResults=500, pageToken=token, projection='full',
fields='nextPageToken,users(primaryEmail,name,orgUnitPath,organizations,phones,suspended,archived)')
if ou != '/':
params['query'] = "orgUnitPath='" + ou.replace("\\", "\\\\").replace("'", "\\'") + "'"
resp = api.users().list(**params).execute()
except Exception as e:
raise HandtekeningFout(f"Gebruikers van {ou} ophalen mislukt: {vriendelijk(e)}")
for u in resp.get('users', []):
pad = u.get('orgUnitPath') or '/'
if not met_sub and pad != ou:
continue
if u.get('suspended') or u.get('archived'):
continue
email = u['primaryEmail'].lower()
if email in gezien:
continue
gezien.add(email)
n = u.get('name') or {}
uit.append({'email': email, 'voornaam': n.get('givenName', ''), 'achternaam': n.get('familyName', ''),
'ou': pad, 'functie': _functie(u), 'telefoon': _telefoon(u)})
token = resp.get('nextPageToken')
if not token:
break
return sorted(uit, key=lambda x: (x['achternaam'].lower(), x['voornaam'].lower()))
def bewaar_in_google(email, functie, telefoon):
"""Functie (primaire organisatie) en werktelefoon van een gebruiker in Google aanpassen; de rest blijft."""
api = _directory()
try:
u = api.users().get(userKey=email, projection='full', fields='organizations,phones').execute()
orgs = u.get('organizations') or []
org = next((o for o in orgs if o.get('primary')), orgs[0] if orgs else None)
if org is None:
if functie:
orgs = [{'title': functie, 'primary': True, 'customType': ''}]
else:
org['title'] = functie
tel = [p for p in (u.get('phones') or []) if p.get('type') != 'work']
if telefoon:
tel.insert(0, {'value': telefoon, 'type': 'work', 'primary': not any(p.get('primary') for p in tel)})
api.users().patch(userKey=email, body={'organizations': orgs, 'phones': tel or []}).execute()
except Exception as e:
raise HandtekeningFout(f"{email}: {vriendelijk(e)}")
def pas_toe(gebruiker, inst, door=None, alleen_gewijzigd=False):
"""Zet de handtekening voor het hoofdadres en de aliassen. Geeft 'ok' / 'ongewijzigd' of gooit HandtekeningFout."""
tekst = vul_in(inst.get('Sjabloon') or '', gegevens(gebruiker), inst)
if len(tekst) > MAX_TEKENS:
raise HandtekeningFout(f"De handtekening is te lang voor Gmail ({len(tekst)} tekens, max. {MAX_TEKENS}).")
h = hashlib.sha1(tekst.encode('utf-8')).hexdigest()
email = gebruiker['email']
if alleen_gewijzigd:
oud = execute_query("SELECT Hash, Ok FROM gmail_handtekening_log WHERE Email = %s", (email,))
if oud and oud[0]['Ok'] and oud[0]['Hash'] == h and _huidige_in_gmail_gelijk(email, tekst):
return 'ongewijzigd'
try:
g = _gmail(email)
adressen = g.users().settings().sendAs().list(userId='me').execute().get('sendAs', [])
for a in adressen:
if a.get('isPrimary') or a.get('verificationStatus') == 'accepted' or a.get('treatAsAlias'):
g.users().settings().sendAs().patch(userId='me', sendAsEmail=a['sendAsEmail'],
body={'signature': tekst}).execute()
melding, ok = None, 1
except Exception as e:
melding, ok = vriendelijk(e), 0
execute_query("INSERT INTO gmail_handtekening_log (Email, Toegepast, Ok, Melding, Hash, Door) "
"VALUES (%s, NOW(), %s, %s, %s, %s) ON DUPLICATE KEY UPDATE Toegepast = NOW(), Ok = VALUES(Ok), "
"Melding = VALUES(Melding), Hash = VALUES(Hash), Door = VALUES(Door)",
(email, ok, melding, h if ok else None, door), commit=True)
if not ok:
raise HandtekeningFout(melding)
return 'ok'
def _huidige_in_gmail_gelijk(email, tekst):
"""Staat de handtekening nog ongewijzigd in Gmail? (De gebruiker kan ze zelf aangepast hebben.)"""
try:
for a in _gmail(email).users().settings().sendAs().list(userId='me').execute().get('sendAs', []):
if a.get('isPrimary'):
gewoon = lambda x: re.sub(r'\s+', '', html.unescape(x or '')).lower()
return gewoon(a.get('signature')) == gewoon(tekst)
except Exception:
pass
return False
# ---------------------------------------------------------------------
# Toepassen op de achtergrond (met voortgang)
# ---------------------------------------------------------------------
class Toepasser:
def __new__(cls):
def init(inst):
inst._lock = threading.Lock()
inst.status = {'bezig': False, 'voortgang': 0, 'totaal': 0, 'gelukt': 0, 'ongewijzigd': 0, 'fouten': 0,
'fout': None, 'resultaat': None, 'foutvoorbeelden': []}
return school_singleton(cls, init)
def run(self, emails=None, door=None, alleen_gewijzigd=False):
if not self._lock.acquire(blocking=False):
return False, "Er loopt al een opdracht."
s = self.status
try:
s.update(bezig=True, voortgang=0, totaal=0, gelukt=0, ongewijzigd=0, fouten=0, fout=None, resultaat=None,
foutvoorbeelden=[])
inst = instellingen()
lijst = gebruikers(inst['ous'], bool(inst.get('Met_sub_ous')))
if emails is not None:
kies = {e.lower() for e in emails}
lijst = [u for u in lijst if u['email'] in kies]
s['totaal'] = len(lijst)
for u in lijst:
try:
if pas_toe(u, inst, door, alleen_gewijzigd) == 'ongewijzigd':
s['ongewijzigd'] += 1
else:
s['gelukt'] += 1
except Exception as e:
s['fouten'] += 1
if len(s['foutvoorbeelden']) < 6:
s['foutvoorbeelden'].append(f"{u['email']}: {str(e)[:180]}")
s['voortgang'] += 1
msg = f"Handtekening ingesteld voor {s['gelukt']} gebruiker(s)"
if s['ongewijzigd']:
msg += f", {s['ongewijzigd']} waren al in orde"
if s['fouten']:
msg += f", {s['fouten']} mislukt"
s['resultaat'] = msg + '.'
return True, s['resultaat']
except Exception as e:
s['fout'] = str(e)[:400]
return False, s['fout']
finally:
s['bezig'] = False
self._lock.release()
def start(self, emails=None, door=None):
if self.status['bezig']:
return False
self.status.update(bezig=True, voortgang=0, totaal=0, fout=None, resultaat=None)
threading.Thread(target=met_school(lambda: self.run(emails, door)), daemon=True).start()
return True
# ---------------------------------------------------------------------
# Nachtelijk opnieuw toepassen (per school met de schakelaar aan)
# ---------------------------------------------------------------------
def _nacht_claim(vandaag):
return execute_query("UPDATE gmail_handtekening SET Laatste_nacht = %s WHERE ID = 1 AND Nachtelijk = 1 "
"AND (Laatste_nacht IS NULL OR Laatste_nacht < %s)", (vandaag, vandaag), commit=True) == 1
def nacht_voor_alle_scholen():
from tenancy import alle_scholen, in_school
vandaag = datetime.now().strftime('%Y-%m-%d')
for school in alle_scholen():
if not school.google_sleutel:
continue
with in_school(school):
try:
ensure_tables()
if not _nacht_claim(vandaag):
continue
ok, msg = Toepasser().run(door='PIVOT (nachtelijk)', alleen_gewijzigd=True)
execute_query("UPDATE gmail_handtekening SET Nacht_resultaat = %s WHERE ID = 1",
(f"{datetime.now():%d-%m-%Y %H:%M}: {msg}"[:500],), commit=True)
except Exception as e:
sys.stderr.write(f"Handtekeningen nachtelijk ({school.code}): {e}\n")
_thread = None
def start_scheduler():
global _thread
if _thread:
return
def loop():
time.sleep(90)
while True:
try:
if NACHT_UUR_VAN <= datetime.now().hour < NACHT_UUR_TOT:
nacht_voor_alle_scholen()
except Exception as e:
sys.stderr.write(f"Handtekeningen nachtelijke taak: {e}\n")
time.sleep(15 * 60)
_thread = threading.Thread(target=loop, daemon=True)
_thread.start()