Christian Mueller
5267baf0b4
Replace the standalone Python script + config.cfg with a full web application featuring: - FastAPI backend with REST API - SQLite database (config, state, activity log separated) - Bootstrap 5 web interface for configuration - Pages: Dashboard, Users, Devices, Assignments, Color Rules, Settings, Log - APScheduler background polling for Teams presence + AIDA status - systemd service file for Linux deployment - Migration script to import existing config.cfg data Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
import logging
|
|
from azure.identity import (
|
|
DeviceCodeCredential,
|
|
TokenCachePersistenceOptions,
|
|
AuthenticationRecord,
|
|
)
|
|
from msgraph.core import GraphClient
|
|
from database import get_setting, set_setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CACHE_OPTIONS = TokenCachePersistenceOptions(
|
|
name="busylight", allow_unencrypted_cache=True
|
|
)
|
|
|
|
|
|
class GraphAPI:
|
|
def __init__(self):
|
|
self.client = None
|
|
self.credential = None
|
|
self._init_client()
|
|
|
|
def _init_client(self):
|
|
client_id = get_setting("azure_client_id")
|
|
tenant_id = get_setting("azure_auth_tenant", "common")
|
|
scopes = get_setting("azure_scopes").split()
|
|
auth_record_json = get_setting("azure_auth_record")
|
|
|
|
if not client_id or not auth_record_json:
|
|
logger.warning("Azure nicht konfiguriert oder nicht authentifiziert")
|
|
return
|
|
|
|
try:
|
|
record = AuthenticationRecord.deserialize(auth_record_json)
|
|
self.credential = DeviceCodeCredential(
|
|
client_id=client_id,
|
|
tenant_id=tenant_id,
|
|
enable_persistent_cache=True,
|
|
allow_unencrypted_cache=True,
|
|
cache_persistence_options=CACHE_OPTIONS,
|
|
authentication_record=record,
|
|
)
|
|
self.credential.authenticate(tenant_id=tenant_id, scopes=scopes)
|
|
self.client = GraphClient(credential=self.credential, scopes=scopes)
|
|
logger.info("Graph-Client erfolgreich initialisiert")
|
|
except Exception as e:
|
|
logger.error("Graph-Client Fehler: %s", e)
|
|
self.client = None
|
|
|
|
def authenticate(self):
|
|
"""Trigger device code flow. Must run in terminal."""
|
|
client_id = get_setting("azure_client_id")
|
|
tenant_id = get_setting("azure_auth_tenant", "common")
|
|
scopes = get_setting("azure_scopes").split()
|
|
|
|
if not client_id:
|
|
raise ValueError("Azure Client ID nicht konfiguriert")
|
|
|
|
self.credential = DeviceCodeCredential(
|
|
client_id=client_id,
|
|
tenant_id=tenant_id,
|
|
enable_persistent_cache=True,
|
|
allow_unencrypted_cache=True,
|
|
cache_persistence_options=CACHE_OPTIONS,
|
|
)
|
|
record = self.credential.authenticate(tenant_id=tenant_id, scopes=scopes)
|
|
set_setting("azure_auth_record", record.serialize())
|
|
self.client = GraphClient(credential=self.credential, scopes=scopes)
|
|
logger.info("Azure Authentifizierung erfolgreich")
|
|
return record
|
|
|
|
def is_authenticated(self):
|
|
return self.client is not None
|
|
|
|
def get_users(self):
|
|
if not self.client:
|
|
return []
|
|
try:
|
|
resp = self.client.get(
|
|
"/users?$select=displayName,id,mail&$orderBy=displayName&$top=999"
|
|
)
|
|
data = resp.json()
|
|
users = data.get("value", [])
|
|
# Handle pagination
|
|
while "@odata.nextLink" in data:
|
|
resp = self.client.get(data["@odata.nextLink"])
|
|
data = resp.json()
|
|
users.extend(data.get("value", []))
|
|
return users
|
|
except Exception as e:
|
|
logger.error("Fehler beim Abrufen der User: %s", e)
|
|
return []
|
|
|
|
def get_presence(self, user_id):
|
|
if not self.client:
|
|
return None
|
|
try:
|
|
resp = self.client.get(f"/users/{user_id}/presence")
|
|
return resp.json()
|
|
except Exception as e:
|
|
logger.error("Fehler beim Abrufen der Presence für %s: %s", user_id, e)
|
|
return None
|