# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Teamsbot Feature Container - Main Module. Handles feature initialization and RBAC catalog registration. """ import logging from typing import Dict, List, Any logger = logging.getLogger(__name__) # Feature metadata FEATURE_CODE = "teamsbot" FEATURE_LABEL = {"en": "Teams Bot", "de": "Teams Bot", "fr": "Teams Bot"} FEATURE_ICON = "mdi-headset" # UI Objects for RBAC catalog UI_OBJECTS = [ { "objectKey": "ui.feature.teamsbot.dashboard", "label": {"en": "Dashboard", "de": "Dashboard", "fr": "Tableau de bord"}, "meta": {"area": "dashboard"} }, { "objectKey": "ui.feature.teamsbot.sessions", "label": {"en": "Sessions", "de": "Sitzungen", "fr": "Sessions"}, "meta": {"area": "sessions"} }, { "objectKey": "ui.feature.teamsbot.settings", "label": {"en": "Settings", "de": "Einstellungen", "fr": "Paramètres"}, "meta": {"area": "settings", "admin_only": True} }, ] # DATA Objects for RBAC catalog (tables/entities) DATA_OBJECTS = [ { "objectKey": "data.feature.teamsbot.TeamsbotSession", "label": {"en": "Session", "de": "Sitzung", "fr": "Session"}, "meta": {"table": "TeamsbotSession", "fields": ["id", "meetingLink", "botName", "status", "startedAt", "endedAt"]} }, { "objectKey": "data.feature.teamsbot.TeamsbotTranscript", "label": {"en": "Transcript", "de": "Transkript", "fr": "Transcription"}, "meta": {"table": "TeamsbotTranscript", "fields": ["id", "sessionId", "speaker", "text", "timestamp"]} }, { "objectKey": "data.feature.teamsbot.TeamsbotBotResponse", "label": {"en": "Bot Response", "de": "Bot-Antwort", "fr": "Réponse du bot"}, "meta": {"table": "TeamsbotBotResponse", "fields": ["id", "sessionId", "responseText", "detectedIntent"]} }, { "objectKey": "data.feature.teamsbot.*", "label": {"en": "All Teams Bot Data", "de": "Alle Teams Bot Daten", "fr": "Toutes les données Teams Bot"}, "meta": {"wildcard": True, "description": "Wildcard for all teamsbot data tables"} }, ] # Resource Objects for RBAC catalog RESOURCE_OBJECTS = [ { "objectKey": "resource.feature.teamsbot.session.start", "label": {"en": "Start Session", "de": "Sitzung starten", "fr": "Démarrer session"}, "meta": {"endpoint": "/api/teamsbot/{instanceId}/sessions", "method": "POST"} }, { "objectKey": "resource.feature.teamsbot.session.stop", "label": {"en": "Stop Session", "de": "Sitzung beenden", "fr": "Arrêter session"}, "meta": {"endpoint": "/api/teamsbot/{instanceId}/sessions/{sessionId}/stop", "method": "POST"} }, { "objectKey": "resource.feature.teamsbot.session.delete", "label": {"en": "Delete Session", "de": "Sitzung löschen", "fr": "Supprimer session"}, "meta": {"endpoint": "/api/teamsbot/{instanceId}/sessions/{sessionId}", "method": "DELETE"} }, { "objectKey": "resource.feature.teamsbot.config.edit", "label": {"en": "Edit Configuration", "de": "Konfiguration bearbeiten", "fr": "Modifier configuration"}, "meta": {"endpoint": "/api/teamsbot/{instanceId}/config", "method": "PUT", "admin_only": True} }, ] # Template roles for this feature with AccessRules TEMPLATE_ROLES = [ { "roleLabel": "teamsbot-admin", "description": { "en": "Teams Bot Administrator - Full access to all sessions and settings", "de": "Teams Bot Administrator - Vollzugriff auf alle Sitzungen und Einstellungen", "fr": "Administrateur Teams Bot - Accès complet aux sessions et paramètres" }, "accessRules": [ # Full UI access (all views including settings) {"context": "UI", "item": None, "view": True}, # Full DATA access {"context": "DATA", "item": None, "view": True, "read": "a", "create": "a", "update": "a", "delete": "a"}, # All resources {"context": "RESOURCE", "item": "resource.feature.teamsbot.session.start", "view": True}, {"context": "RESOURCE", "item": "resource.feature.teamsbot.session.stop", "view": True}, {"context": "RESOURCE", "item": "resource.feature.teamsbot.session.delete", "view": True}, {"context": "RESOURCE", "item": "resource.feature.teamsbot.config.edit", "view": True}, ] }, { "roleLabel": "teamsbot-viewer", "description": { "en": "Teams Bot Viewer - View sessions and transcripts (read-only)", "de": "Teams Bot Betrachter - Sitzungen und Transkripte ansehen (nur lesen)", "fr": "Visualiseur Teams Bot - Consulter les sessions et transcriptions (lecture seule)", }, "accessRules": [ {"context": "UI", "item": "ui.feature.teamsbot.dashboard", "view": True}, {"context": "UI", "item": "ui.feature.teamsbot.sessions", "view": True}, {"context": "DATA", "item": None, "view": True, "read": "m", "create": "n", "update": "n", "delete": "n"}, ], }, { "roleLabel": "teamsbot-user", "description": { "en": "Teams Bot User - Can start/stop sessions and view transcripts", "de": "Teams Bot Benutzer - Kann Sitzungen starten/stoppen und Transkripte einsehen", "fr": "Utilisateur Teams Bot - Peut démarrer/arrêter des sessions et voir les transcriptions", }, "accessRules": [ {"context": "UI", "item": "ui.feature.teamsbot.dashboard", "view": True}, {"context": "UI", "item": "ui.feature.teamsbot.sessions", "view": True}, {"context": "DATA", "item": "data.feature.teamsbot.TeamsbotSession", "view": True, "read": "m", "create": "m", "update": "m", "delete": "n"}, {"context": "DATA", "item": "data.feature.teamsbot.TeamsbotTranscript", "view": True, "read": "m", "create": "n", "update": "n", "delete": "n"}, {"context": "DATA", "item": "data.feature.teamsbot.TeamsbotBotResponse", "view": True, "read": "m", "create": "n", "update": "n", "delete": "n"}, {"context": "RESOURCE", "item": "resource.feature.teamsbot.session.start", "view": True}, {"context": "RESOURCE", "item": "resource.feature.teamsbot.session.stop", "view": True}, ], }, ] def getFeatureDefinition() -> Dict[str, Any]: """Return the feature definition for registration.""" return { "code": FEATURE_CODE, "label": FEATURE_LABEL, "icon": FEATURE_ICON, "autoCreateInstance": False, } def getUiObjects() -> List[Dict[str, Any]]: """Return UI objects for RBAC catalog registration.""" return UI_OBJECTS def getResourceObjects() -> List[Dict[str, Any]]: """Return resource objects for RBAC catalog registration.""" return RESOURCE_OBJECTS def getTemplateRoles() -> List[Dict[str, Any]]: """Return template roles for this feature.""" return TEMPLATE_ROLES def getDataObjects() -> List[Dict[str, Any]]: """Return DATA objects for RBAC catalog registration.""" return DATA_OBJECTS def registerFeature(catalogService) -> bool: """Register this feature's RBAC objects in the catalog.""" try: for uiObj in UI_OBJECTS: catalogService.registerUiObject( featureCode=FEATURE_CODE, objectKey=uiObj["objectKey"], label=uiObj["label"], meta=uiObj.get("meta") ) for resObj in RESOURCE_OBJECTS: catalogService.registerResourceObject( featureCode=FEATURE_CODE, objectKey=resObj["objectKey"], label=resObj["label"], meta=resObj.get("meta") ) for dataObj in DATA_OBJECTS: catalogService.registerDataObject( featureCode=FEATURE_CODE, objectKey=dataObj["objectKey"], label=dataObj["label"], meta=dataObj.get("meta") ) _syncTemplateRolesToDb() logger.info(f"Feature '{FEATURE_CODE}' registered {len(UI_OBJECTS)} UI, {len(RESOURCE_OBJECTS)} resource, {len(DATA_OBJECTS)} data objects") return True except Exception as e: logger.error(f"Failed to register feature '{FEATURE_CODE}': {e}") return False def _syncTemplateRolesToDb() -> int: """Sync template roles and their AccessRules to the database.""" try: from modules.interfaces.interfaceDbApp import getRootInterface from modules.datamodels.datamodelRbac import Role, AccessRule, AccessRuleContext rootInterface = getRootInterface() existingRoles = rootInterface.getRolesByFeatureCode(FEATURE_CODE) templateRoles = [r for r in existingRoles if r.mandateId is None] existingRoleLabels = {r.roleLabel: str(r.id) for r in templateRoles} createdCount = 0 for roleTemplate in TEMPLATE_ROLES: roleLabel = roleTemplate["roleLabel"] if roleLabel in existingRoleLabels: roleId = existingRoleLabels[roleLabel] _ensureAccessRulesForRole(rootInterface, roleId, roleTemplate.get("accessRules", [])) else: newRole = Role( roleLabel=roleLabel, description=roleTemplate.get("description", {}), featureCode=FEATURE_CODE, mandateId=None, featureInstanceId=None, isSystemRole=False ) createdRole = rootInterface.db.recordCreate(Role, newRole.model_dump()) roleId = createdRole.get("id") _ensureAccessRulesForRole(rootInterface, roleId, roleTemplate.get("accessRules", [])) logger.info(f"Created template role '{roleLabel}' with ID {roleId}") createdCount += 1 if createdCount > 0: logger.info(f"Feature '{FEATURE_CODE}': Created {createdCount} template roles") return createdCount except Exception as e: logger.error(f"Error syncing template roles for feature '{FEATURE_CODE}': {e}") return 0 def _ensureAccessRulesForRole(rootInterface, roleId: str, ruleTemplates: List[Dict[str, Any]]) -> int: """Ensure AccessRules exist for a role based on templates.""" from modules.datamodels.datamodelRbac import AccessRule, AccessRuleContext existingRules = rootInterface.getAccessRulesByRole(roleId) existingSignatures = set() for rule in existingRules: sig = (rule.context.value if rule.context else None, rule.item) existingSignatures.add(sig) createdCount = 0 for template in ruleTemplates: context = template.get("context", "UI") item = template.get("item") sig = (context, item) if sig in existingSignatures: continue if context == "UI": contextEnum = AccessRuleContext.UI elif context == "DATA": contextEnum = AccessRuleContext.DATA elif context == "RESOURCE": contextEnum = AccessRuleContext.RESOURCE else: contextEnum = context newRule = AccessRule( roleId=roleId, context=contextEnum, item=item, view=template.get("view", False), read=template.get("read"), create=template.get("create"), update=template.get("update"), delete=template.get("delete"), ) rootInterface.db.recordCreate(AccessRule, newRule.model_dump()) createdCount += 1 if createdCount > 0: logger.debug(f"Created {createdCount} AccessRules for role {roleId}") return createdCount