import json import uuid from datetime import datetime as dt, timezone as tz from flask import request, redirect, flash, render_template, Blueprint, session, current_app from flask_security import roles_accepted, current_user from sqlalchemy.exc import SQLAlchemyError, IntegrityError import ast from wtforms import BooleanField from common.models.user import User, Tenant, Role, TenantDomain, TenantProject, PartnerTenant, TenantMake, \ ConsentVersion, TenantConsent from common.extensions import db, security, minio_client, simple_encryption, cache_manager, content_manager from common.services.utils.version_services import VersionServices from common.utils.dynamic_field_utils import create_default_config_from_type_config from common.utils.security_utils import send_confirmation_email, send_reset_email from config.type_defs.service_types import SERVICE_TYPES from .user_forms import TenantForm, CreateUserForm, EditUserForm, TenantDomainForm, TenantSelectionForm, \ TenantProjectForm, EditTenantProjectForm, TenantMakeForm, EditTenantForm, EditTenantMakeForm, ConsentVersionForm, \ EditConsentVersionForm from common.utils.database import Database from common.utils.view_assistants import prepare_table_for_macro, form_validation_failed from common.utils.simple_encryption import generate_api_key from common.utils.nginx_utils import prefixed_url_for from common.utils.eveai_exceptions import EveAIException from common.utils.document_utils import set_logging_information, update_logging_information from common.services.user import TenantServices from common.services.user import UserServices from common.utils.mail_utils import send_email from eveai_app.views.list_views.user_list_views import get_tenants_list_view, get_users_list_view, \ get_tenant_domains_list_view, get_tenant_projects_list_view, get_tenant_makes_list_view, \ get_tenant_partner_services_list_view, get_consent_versions_list_view, get_tenant_consents_list_view from eveai_app.views.list_views.list_view_utils import render_list_view from common.services.user.consent_services import ConsentServices from common.models.user import ConsentStatus user_bp = Blueprint('user_bp', __name__, url_prefix='/user') @user_bp.before_request def log_before_request(): current_app.logger.debug(f'Before request: {request.path} =====================================') @user_bp.after_request def log_after_request(response): return response # Tenant Management ------------------------------------------------------------------------------- @user_bp.route('/tenant', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin') def tenant(): if not UserServices.can_user_create_tenant(): current_app.logger.error(f'User {current_user.email} cannot create tenant') flash(f"You don't have the appropriate permissions to create a tenant", 'danger') return redirect(prefixed_url_for('user_bp.tenants', for_redirect=True)) form = TenantForm() if request.method == 'GET': code = f"TENANT-{str(uuid.uuid4())}" form.code.data = code if form.validate_on_submit(): # Handle the required attributes new_tenant = Tenant() form.populate_obj(new_tenant) timestamp = dt.now(tz.utc) new_tenant.created_at = timestamp new_tenant.updated_at = timestamp # Add the new tenant to the database and commit the changes try: db.session.add(new_tenant) db.session.commit() if current_user.has_roles('Partner Admin') and 'partner' in session: # Always associate with the partner for Partner Admins TenantServices.associate_tenant_with_partner(new_tenant.id) elif current_user.has_roles('Super User') and form.assign_to_partner.data and 'partner' in session: # Super User chose to associate with partner TenantServices.associate_tenant_with_partner(new_tenant.id) except IntegrityError as e: db.session.rollback() # Check for the specific error about duplicate tenant name if "tenant_name_key" in str(e) or "duplicate key value" in str(e): flash(f"A tenant with the name '{form.name.data}' already exists. Please choose a different name.", 'danger') else: current_app.logger.error(f'Failed to add tenant to database. Error: {str(e)}') flash(f'Failed to add tenant to database. Error: {str(e)}', 'danger') return render_template('user/tenant.html', form=form) except SQLAlchemyError as e: current_app.logger.error(f'Failed to add tenant to database. Error: {str(e)}') flash(f'Failed to add tenant to database. Error: {str(e)}', 'danger') return render_template('user/tenant.html', form=form) except EveAIException as e: current_app.logger.error(f'Error associating Tenant {new_tenant.id} to Partner. Error: {str(e)}') flash(f'Error associating Tenant to Partner. Error: {str(e)}', 'danger') return render_template('user/tenant.html', form=form) current_app.logger.info(f"Successfully created tenant {new_tenant.id} in Database") flash(f"Successfully created tenant {new_tenant.id} in Database", 'success') # Create schema for new tenant current_app.logger.info(f"Creating schema for tenant {new_tenant.id}") Database(new_tenant.id).create_tenant_schema() # Create MinIO bucket for new tenant current_app.logger.info(f"Creating MinIO bucket for tenant {new_tenant.id}") minio_client.create_tenant_bucket(new_tenant.id) return redirect(prefixed_url_for('user_bp.tenants', for_redirect=True)) else: form_validation_failed(request, form) return render_template('user/tenant.html', form=form) @user_bp.route('/tenant/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin') def edit_tenant(tenant_id): tenant = Tenant.query.get_or_404(tenant_id) # This will return a 404 if no tenant is found form = EditTenantForm(obj=tenant) if form.validate_on_submit(): # Populate the tenant with form data form.populate_obj(tenant) # Convert default_tenant_make_id to integer if not empty if form.default_tenant_make_id.data: tenant.default_tenant_make_id = int(form.default_tenant_make_id.data) else: tenant.default_tenant_make_id = None db.session.commit() flash('Tenant updated successfully.', 'success') if session.get('tenant'): if session['tenant'].get('id') == tenant_id: session['tenant'] = tenant.to_dict() # return redirect(url_for(f"user/tenant/tenant_id")) else: form_validation_failed(request, form) return render_template('user/tenant.html', form=form, tenant_id=tenant_id) @user_bp.route('/tenants', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin') # Allow both roles def tenants(): # Get configuration and render the list view config = get_tenants_list_view() return render_list_view('list_view.html', **config) @user_bp.route('/handle_tenant_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin') def handle_tenant_selection(): action = request.form['action'] if action == 'create_tenant': return redirect(prefixed_url_for('user_bp.tenant', for_redirect=True)) tenant_identification = request.form['selected_row'] tenant_id = ast.literal_eval(tenant_identification).get('value') if not UserServices.can_user_edit_tenant(tenant_id): current_app.logger.info(f"User not authenticated to edit tenant {tenant_id}.") flash(f"You are not authenticated to manage tenant {tenant_id}", 'danger') return redirect(prefixed_url_for('user_bp.tenants', for_redirect=True)) the_tenant = Tenant.query.get(tenant_id) # set tenant information in the session session['tenant'] = the_tenant.to_dict() # remove catalog-related items from the session session.pop('catalog_id', None) session.pop('catalog_name', None) match action: case 'edit_tenant': return redirect(prefixed_url_for('user_bp.edit_tenant', tenant_id=tenant_id, for_redirect=True)) case 'select_tenant': return redirect(prefixed_url_for('user_bp.tenant_overview', for_redirect=True)) # Add more conditions for other actions return redirect(prefixed_url_for('tenants', for_redirect=True)) @user_bp.route('/tenant_overview', methods=['GET']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_overview(): tenant_id = session['tenant']['id'] tenant = Tenant.query.get_or_404(tenant_id) form = EditTenantForm(obj=tenant) # Zet de waarde van default_tenant_make_id if tenant.default_tenant_make_id: form.default_tenant_make_id.data = str(tenant.default_tenant_make_id) # Haal de naam van de default make op als deze bestaat default_make_name = None if tenant.default_tenant_make: default_make_name = tenant.default_tenant_make.name return render_template('user/tenant_overview.html', form=form, default_make_name=default_make_name) # User Management --------------------------------------------------------------------------------- @user_bp.route('/user', methods=['GET', 'POST']) @roles_accepted('Super User', 'Tenant Admin', 'Partner Admin') def user(): tenant_id = session.get('tenant').get('id') form = CreateUserForm() form.tenant_id.data = session.get('tenant').get('id') # It is only possible to create users for the session tenant if form.validate_on_submit(): current_app.logger.info(f"Adding User for tenant {session['tenant']['id']} ") new_user = User(user_name=form.user_name.data, email=form.email.data, first_name=form.first_name.data, last_name=form.last_name.data, valid_to=form.valid_to.data, tenant_id=form.tenant_id.data, fs_uniquifier=uuid.uuid4().hex, ) timestamp = dt.now(tz.utc) new_user.created_at = timestamp new_user.updated_at = timestamp # Add roles for role_id in form.roles.data: the_role = Role.query.get(role_id) new_user.roles.append(the_role) # Add the new user to the database and commit the changes try: db.session.add(new_user) db.session.commit() # security.datastore.set_uniquifier(new_user) try: send_confirmation_email(new_user) current_app.logger.info(f'User {new_user.id} with name {new_user.user_name} added to database' f'Confirmation email sent to {new_user.email}') flash('User added successfully and confirmation email sent.', 'success') except Exception as e: current_app.logger.error(f'Failed to send confirmation email to {new_user.email}. Error: {str(e)}') flash('User added successfully, but failed to send confirmation email. ' 'Please contact the administrator.', 'warning') return redirect(prefixed_url_for('user_bp.view_users', for_redirect=True)) except Exception as e: current_app.logger.error(f'Failed to add user with name {new_user.user_name}. Error: {str(e)}') db.session.rollback() flash(f'Failed to add user. Email or user name already exists.', 'danger') else: form_validation_failed(request, form) return render_template('user/user.html', form=form) @user_bp.route('/user/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Tenant Admin', 'Partner Admin') def edit_user(user_id): user = User.query.get_or_404(user_id) # This will return a 404 if no user is found form = EditUserForm(obj=user) if form.validate_on_submit(): # Populate the user with form data user.first_name = form.first_name.data user.last_name = form.last_name.data user.valid_to = form.valid_to.data user.updated_at = dt.now(tz.utc) # Update roles current_roles = set(role.id for role in user.roles) selected_roles = set(form.roles.data) if UserServices.validate_role_assignments(selected_roles): # Add new roles for role_id in selected_roles - current_roles: role = Role.query.get(role_id) if role: user.roles.append(role) # Remove unselected roles for role_id in current_roles - selected_roles: role = Role.query.get(role_id) if role: user.roles.remove(role) else: flash('Trying to assign unauthorized roles', 'danger') current_app.logger.error(f"Trying to assign unauthorized roles by user {user_id}," f"tenant {session['tenant']['id']}") return redirect(prefixed_url_for('user_bp.edit_user', user_id=user_id, for_redirect=True)) db.session.commit() flash('User updated successfully.', 'success') return redirect( prefixed_url_for('user_bp.edit_user', user_id=user.id, for_redirect=True)) # Assuming there's a user profile view to redirect to else: form_validation_failed(request, form) form.roles.data = [role.id for role in user.roles] return render_template('user/edit_user.html', form=form, user_id=user_id) @user_bp.route('/view_users') @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def view_users(): tenant_id = session.get('tenant').get('id') config = get_users_list_view(tenant_id) return render_list_view('list_view.html', **config) @user_bp.route('/handle_user_action', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_user_action(): action = request.form['action'] if action == 'create_user': return redirect(prefixed_url_for('user_bp.user', for_redirect=True)) user_identification = request.form['selected_row'] user_id = ast.literal_eval(user_identification).get('value') user = User.query.get_or_404(user_id) if action == 'edit_user': return redirect(prefixed_url_for('user_bp.edit_user', user_id=user_id, for_redirect=True)) elif action == 'resend_confirmation_email': send_confirmation_email(user) flash(f'Confirmation email sent to {user.email}.', 'success') elif action == 'send_password_reset_email': send_reset_email(user) flash(f'Password reset email sent to {user.email}.', 'success') elif action == 'reset_uniquifier': reset_uniquifier(user) flash(f'Uniquifier reset for {user.user_name}.', 'success') return redirect(prefixed_url_for('user_bp.view_users', for_redirect=True)) # Tenant Domain Management (Probably obsolete )------------------------------------------------------------------------ @user_bp.route('/tenant_domains') @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_domains(): tenant_id = session['tenant']['id'] config = get_tenant_domains_list_view(tenant_id) return render_list_view('list_view.html', **config) @user_bp.route('/handle_tenant_domain_action', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_tenant_domain_action(): action = request.form['action'] if action == 'create_tenant_domain': return redirect(prefixed_url_for('user_bp.tenant_domain', for_redirect=True)) tenant_domain_identification = request.form['selected_row'] tenant_domain_id = ast.literal_eval(tenant_domain_identification).get('value') if action == 'edit_tenant_domain': return redirect(prefixed_url_for('user_bp.edit_tenant_domain', tenant_domain_id=tenant_domain_id, for_redirect=True)) # Add more conditions for other actions return redirect(prefixed_url_for('tenant_domains', for_redirect=True)) @user_bp.route('/tenant_domain', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_domain(): form = TenantDomainForm() if form.validate_on_submit(): new_tenant_domain = TenantDomain() form.populate_obj(new_tenant_domain) new_tenant_domain.tenant_id = session['tenant']['id'] set_logging_information(new_tenant_domain, dt.now(tz.utc)) # Add the new user to the database and commit the changes try: db.session.add(new_tenant_domain) db.session.commit() flash('Tenant Domain added successfully.', 'success') current_app.logger.info( f'Tenant Domain {new_tenant_domain.domain} added for tenant {session["tenant"]["id"]}') except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to add Tenant Domain. Error: {str(e)}', 'danger') current_app.logger.error(f'Failed to create Tenant Domain {new_tenant_domain.domain}. ' f'for tenant {session["tenant"]["id"]}' f'Error: {str(e)}') else: flash('Please fill in all required fields.', 'information') return render_template('user/tenant_domain.html', form=form) @user_bp.route('/tenant_domain/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def edit_tenant_domain(tenant_domain_id): tenant_domain = TenantDomain.query.get_or_404(tenant_domain_id) # This will return a 404 if no user is found form = TenantDomainForm(obj=tenant_domain) if request.method == 'POST' and form.validate_on_submit(): form.populate_obj(tenant_domain) update_logging_information(tenant_domain, dt.now(tz.utc)) try: db.session.add(tenant_domain) db.session.commit() flash('Tenant Domain updated successfully.', 'success') except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to update Tenant Domain. Error: {str(e)}', 'danger') current_app.logger.error(f'Failed to update Tenant Domain {tenant_domain.id}. ' f'for tenant {session["tenant"]["id"]}' f'Error: {str(e)}') return redirect( prefixed_url_for('user_bp.tenant_domains', tenant_id=session['tenant']['id'], for_redirect=True)) # Assuming there's a user profile view to redirect to else: form_validation_failed(request, form) return render_template('user/edit_tenant_domain.html', form=form, tenant_domain_id=tenant_domain_id) # Tenant Project Management ----------------------------------------------------------------------- @user_bp.route('/tenant_project', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_project(): form = TenantProjectForm() if request.method == 'GET': # Initialize the API key new_api_key = generate_api_key(prefix="EveAI") form.unencrypted_api_key.data = new_api_key form.visual_api_key.data = f"EVEAI-...{new_api_key[-4:]}" if form.validate_on_submit(): new_tenant_project = TenantProject() form.populate_obj(new_tenant_project) new_tenant_project.tenant_id = session['tenant']['id'] new_tenant_project.encrypted_api_key = simple_encryption.encrypt_api_key(new_tenant_project.unencrypted_api_key) set_logging_information(new_tenant_project, dt.now(tz.utc)) # Add new Tenant Project to the database try: db.session.add(new_tenant_project) db.session.commit() # Send email notification services = [SERVICE_TYPES[service]['name'] for service in form.services.data if service in SERVICE_TYPES] email_sent = send_api_key_notification( tenant_id=session['tenant']['id'], tenant_name=session['tenant']['name'], project_name=new_tenant_project.name, api_key=new_tenant_project.unencrypted_api_key, services=services, responsible_email=form.responsible_email.data ) if email_sent: flash('Tenant Project created successfully and notification email sent.', 'success') else: flash('Tenant Project created successfully but failed to send notification email.', 'warning') current_app.logger.info(f'Tenant Project {new_tenant_project.name} added for tenant ' f'{session['tenant']['id']}.') return redirect(prefixed_url_for('user_bp.tenant_projects', for_redirect=True)) except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to create Tenant Project. Error: {str(e)}', 'danger') current_app.logger.error(f"Failed to create Tenant Project for tenant {session['tenant']['id']}. " f"Error: {str(e)}") return render_template('user/tenant_project.html', form=form) @user_bp.route('/tenant_projects', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_projects(): tenant_id = session['tenant']['id'] config = get_tenant_projects_list_view(tenant_id) return render_list_view('list_view.html', **config) @user_bp.route('/handle_tenant_project_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_tenant_project_selection(): action = request.form.get('action') if action == 'create_tenant_project': return redirect(prefixed_url_for('user_bp.tenant_project', for_redirect=True)) tenant_project_identification = request.form.get('selected_row') tenant_project_id = ast.literal_eval(tenant_project_identification).get('value') tenant_project = TenantProject.query.get_or_404(tenant_project_id) if action == 'edit_tenant_project': return redirect(prefixed_url_for('user_bp.edit_tenant_project', tenant_project_id=tenant_project_id, for_redirect=True)) elif action == 'invalidate_tenant_project': tenant_project.active = False try: db.session.add(tenant_project) db.session.commit() flash('Tenant Project invalidated successfully.', 'success') current_app.logger.info(f'Tenant Project {tenant_project.name} invalidated for tenant ' f'{session['tenant']['id']}.') except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to invalidate Tenant Project {tenant_project.name}. Error: {str(e)}', 'danger') current_app.logger.error(f"Failed to invalidate Tenant Project for tenant {session['tenant']['id']}. " f"Error: {str(e)}") elif action == 'delete_tenant_project': return redirect(prefixed_url_for('user_bp.delete_tenant_project', tenant_project_id=tenant_project_id, for_redirect=True)) return redirect(prefixed_url_for('user_bp.tenant_projects', for_redirect=True)) @user_bp.route('/tenant_project/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def edit_tenant_project(tenant_project_id): tenant_project = TenantProject.query.get_or_404(tenant_project_id) tenant_id = session['tenant']['id'] form = EditTenantProjectForm(obj=tenant_project) if form.validate_on_submit(): form.populate_obj(tenant_project) update_logging_information(tenant_project, dt.now(tz.utc)) try: db.session.add(tenant_project) db.session.commit() flash('Tenant Project updated successfully.', 'success') current_app.logger.info(f'Tenant Project {tenant_project.name} updated for tenant {tenant_id}.') return redirect(prefixed_url_for('user_bp.tenant_projects', for_redirect=True)) except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to update Tenant Project. Error: {str(e)}', 'danger') current_app.logger.error(f"Failed to update Tenant Project {tenant_project.name} for tenant {tenant_id}. ") return render_template('user/edit_tenant.html', form=form, tenant_project_id=tenant_project_id) @user_bp.route('/tenant_project/delete/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def delete_tenant_project(tenant_project_id): tenant_id = session['tenant']['id'] tenant_project = TenantProject.query.get_or_404(tenant_project_id) # Ensure project belongs to current tenant if tenant_project.tenant_id != tenant_id: flash('You do not have permission to delete this project.', 'danger') return redirect(prefixed_url_for('user_bp.tenant_projects', for_redirect=True)) if request.method == 'GET': return render_template('user/confirm_delete_tenant_project.html', tenant_project=tenant_project) try: project_name = tenant_project.name db.session.delete(tenant_project) db.session.commit() flash(f'Tenant Project "{project_name}" successfully deleted.', 'success') current_app.logger.info(f'Tenant Project {project_name} deleted for tenant {tenant_id}') except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to delete Tenant Project. Error: {str(e)}', 'danger') current_app.logger.error(f'Failed to delete Tenant Project {tenant_project_id}. Error: {str(e)}') return redirect(prefixed_url_for('user_bp.tenant_projects', for_redirect=True)) # Tenant Make Management -------------------------------------------------------------------------- @user_bp.route('/tenant_make', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_make(): form = TenantMakeForm() customisation_config = cache_manager.customisations_config_cache.get_config("CHAT_CLIENT_CUSTOMISATION") default_customisation_options = create_default_config_from_type_config(customisation_config["configuration"]) if form.validate_on_submit(): tenant_id = session['tenant']['id'] new_tenant_make = TenantMake() form.populate_obj(new_tenant_make) new_tenant_make.tenant_id = tenant_id # Verwerk allowed_languages als array new_tenant_make.allowed_languages = form.allowed_languages.data if form.allowed_languages.data else None set_logging_information(new_tenant_make, dt.now(tz.utc)) try: db.session.add(new_tenant_make) db.session.commit() flash('Tenant Make successfully added!', 'success') current_app.logger.info(f'Tenant Make {new_tenant_make.name}, id {new_tenant_make.id} successfully added ' f'for tenant {tenant_id}!') # Enable step 2 of creation of retriever - add configuration of the retriever (dependent on type) return redirect(prefixed_url_for('user_bp.edit_tenant_make', tenant_make_id=new_tenant_make.id, for_redirect=True)) except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to add Tenant Make. Error: {e}', 'danger') current_app.logger.error(f'Failed to add Tenant Make {new_tenant_make.name}' f'for tenant {tenant_id}. Error: {str(e)}') return render_template('user/tenant_make.html', form=form) @user_bp.route('/tenant_makes', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_makes(): tenant_id = session['tenant']['id'] config = get_tenant_makes_list_view(tenant_id) return render_list_view('list_view.html', **config) @user_bp.route('/tenant_make/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def edit_tenant_make(tenant_make_id): """Edit an existing tenant make configuration.""" # Get the tenant make or return 404 tenant_make = TenantMake.query.get_or_404(tenant_make_id) # Create form instance with the tenant make form = EditTenantMakeForm(request.form, obj=tenant_make) # Initialiseer de allowed_languages selectie met huidige waarden if request.method == 'GET': if tenant_make.allowed_languages: form.allowed_languages.data = tenant_make.allowed_languages customisation_config = cache_manager.customisations_config_cache.get_config("CHAT_CLIENT_CUSTOMISATION") form.add_dynamic_fields("configuration", customisation_config, tenant_make.chat_customisation_options) if form.validate_on_submit(): # Update basic fields form.populate_obj(tenant_make) tenant_make.chat_customisation_options = form.get_dynamic_data("configuration") # Verwerk allowed_languages als array tenant_make.allowed_languages = form.allowed_languages.data if form.allowed_languages.data else None # Update logging information update_logging_information(tenant_make, dt.now(tz.utc)) # Save changes to database try: db.session.add(tenant_make) db.session.commit() flash('Tenant Make updated successfully!', 'success') current_app.logger.info(f'Tenant Make {tenant_make.id} updated successfully') except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to update tenant make. Error: {str(e)}', 'danger') current_app.logger.error(f'Failed to update tenant make {tenant_make_id}. Error: {str(e)}') return render_template('user/edit_tenant_make.html', form=form, tenant_make_id=tenant_make_id) return redirect(prefixed_url_for('user_bp.tenant_makes', for_redirect=True)) else: form_validation_failed(request, form) return render_template('user/edit_tenant_make.html', form=form, tenant_make_id=tenant_make_id) @user_bp.route('/handle_tenant_make_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_tenant_make_selection(): action = request.form['action'] if action == 'create_tenant_make': return redirect(prefixed_url_for('user_bp.tenant_make', for_redirect=True)) tenant_make_identification = request.form.get('selected_row') tenant_make_id = ast.literal_eval(tenant_make_identification).get('value') if action == 'edit_tenant_make': return redirect(prefixed_url_for('user_bp.edit_tenant_make', tenant_make_id=tenant_make_id, for_redirect=True)) elif action == 'set_as_default': # Set this make as the default for the tenant tenant_id = session['tenant']['id'] tenant = Tenant.query.get(tenant_id) tenant.default_tenant_make_id = tenant_make_id try: db.session.commit() flash(f'Default tenant make updated successfully.', 'success') # Update session data if necessary if 'tenant' in session: session['tenant'] = tenant.to_dict() except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to update default tenant make. Error: {str(e)}', 'danger') current_app.logger.error(f'Failed to update default tenant make. Error: {str(e)}') # Altijd teruggaan naar de tenant_makes pagina return redirect(prefixed_url_for('user_bp.tenant_makes', for_redirect=True)) @user_bp.route('/tenant_partner_services', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_partner_services(): tenant_id = session['tenant']['id'] config = get_tenant_partner_services_list_view(tenant_id) return render_list_view('list_view.html', **config) # Consent Version Management ---------------------------------------------------------------------- @user_bp.route('/consent_versions', methods=['GET', 'POST']) @roles_accepted('Super User') def consent_versions(): config = get_consent_versions_list_view() return render_list_view('list_view.html', **config) @user_bp.route('/handle_consent_version_selection', methods=['POST']) @roles_accepted('Super User') def handle_consent_version_selection(): action = request.form['action'] if action == 'create_consent_version': return redirect(prefixed_url_for('user_bp.consent_version', for_redirect=True)) consent_version_identification = request.form.get('selected_row') consent_version_id = ast.literal_eval(consent_version_identification).get('value') if action == 'edit_consent_version': return redirect(prefixed_url_for('user_bp.edit_consent_version', consent_version_id=consent_version_id, for_redirect=True)) # Altijd teruggaan naar de tenant_makes pagina return redirect(prefixed_url_for('user_bp.consent_versions', for_redirect=True)) @user_bp.route('/consent_version', methods=['GET', 'POST']) @roles_accepted('Super User') def consent_version(): form = ConsentVersionForm() if form.validate_on_submit(): new_consent_version = ConsentVersion() form.populate_obj(new_consent_version) set_logging_information(new_consent_version, dt.now(tz.utc)) try: db.session.add(new_consent_version) db.session.commit() flash('Consent Version successfully added!', 'success') current_app.logger.info(f'Consent Version {new_consent_version.consent_type}, version {new_consent_version.consent_version} successfully added ') # Enable step 2 of creation of retriever - add configuration of the retriever (dependent on type) return redirect(prefixed_url_for('user_bp.consent_versions', for_redirect=True)) except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to add Consent Version. Error: {e}', 'danger') current_app.logger.error(f'Failed to add Consent Version. Error: {str(e)}') return render_template('user/consent_version.html', form=form) @user_bp.route('/consent_version/', methods=['GET', 'POST']) @roles_accepted('Super User') def edit_consent_version(consent_version_id): """Edit an existing Consent Version.""" # Get the Consent Version or return 404 cv = ConsentVersion.query.get_or_404(consent_version_id) # Create form instance with the tenant make form = EditConsentVersionForm(request.form, obj=cv) if form.validate_on_submit(): # Update basic fields form.populate_obj(cv) # Update logging information update_logging_information(cv, dt.now(tz.utc)) # Save changes to database try: db.session.add(cv) db.session.commit() flash('Consent Version updated successfully!', 'success') current_app.logger.info(f'Consent Version {cv.id} updated successfully') except SQLAlchemyError as e: db.session.rollback() flash(f'Failed to update Consent Version. Error: {str(e)}', 'danger') current_app.logger.error(f'Failed to update Consent Version {consent_version_id}. Error: {str(e)}') return render_template('user/consent_version.html', form=form, consent_version_id=consent_version_id) return redirect(prefixed_url_for('user_bp.consent_versions', for_redirect=True)) else: form_validation_failed(request, form) return render_template('user/edit_consent_version.html', form=form, consent_version_id=consent_version_id) # Tenant Consent Management ----------------------------------------------------------------------- @user_bp.route('/consent/tenant', methods=['GET']) @roles_accepted('Tenant Admin') def tenant_consent(): # Overview for current session tenant tenant_id = session.get('tenant', {}).get('id') or getattr(current_user, 'tenant_id', None) if not tenant_id: flash('No tenant context.', 'danger') return redirect(prefixed_url_for('user_bp.tenants', for_redirect=True)) types = ConsentServices.get_required_consent_types() statuses = [ConsentServices.evaluate_type_status(tenant_id, t) for t in types] if current_app.jinja_env.loader: return render_template('user/tenant_consent.html', statuses=statuses, tenant_id=tenant_id) # Fallback text if no templates lines = [f"{s.consent_type}: {s.status} (active={s.active_version}, last={s.last_version})" for s in statuses] return "\n".join(lines) @user_bp.route('/consent/no_access', methods=['GET']) def no_consent(): return render_template('user/no_consent.html') if current_app.jinja_env.loader else "Consent required - contact your admin" @user_bp.route('/consent/tenant_renewal', methods=['GET']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_consent_renewal(): # Show renewal statuses only tenant_id = session.get('tenant', {}).get('id') or getattr(current_user, 'tenant_id', None) types = ConsentServices.get_required_consent_types() statuses = [s for s in [ConsentServices.evaluate_type_status(tenant_id, t) for t in types] if s.status != ConsentStatus.CONSENTED] if current_app.jinja_env.loader: return render_template('user/tenant_consent_renewal.html', statuses=statuses, tenant_id=tenant_id) return "\n".join([f"{s.consent_type}: {s.status}" for s in statuses]) @user_bp.route('/consent/renewal', methods=['GET']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def consent_renewal(): return render_template('user/consent_renewal.html') if current_app.jinja_env.loader else "Consent renewal in progress" @user_bp.route('/tenants//consents', methods=['GET']) @roles_accepted('Super User', 'Partner Admin') def view_tenant_consents(tenant_id: int): # Authorization: Tenant Admin for own tenant or Management Partner allowed, mode, _, _ = ConsentServices.can_consent_on_behalf(tenant_id) if not (allowed or current_user.has_roles('Super User')): flash('Not authorized to view consents for this tenant', 'danger') return redirect(prefixed_url_for('user_bp.tenants', for_redirect=True)) types = ConsentServices.get_required_consent_types() statuses = [ConsentServices.evaluate_type_status(tenant_id, t) for t in types] if current_app.jinja_env.loader: return render_template('user/tenant_consents_overview.html', statuses=statuses, tenant_id=tenant_id) return "\n".join([f"{s.consent_type}: {s.status}" for s in statuses]) @user_bp.route('/tenants//consents//accept', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def accept_tenant_consent(tenant_id: int, consent_type: str): try: tc = ConsentServices.record_consent(tenant_id, consent_type) flash(f"Consent for {consent_type} recorded (version {tc.consent_version})", 'success') except PermissionError: flash('Not authorized to accept this consent for the tenant', 'danger') except Exception as e: current_app.logger.error(f"Failed to record consent: {e}") flash('Failed to record consent', 'danger') if current_user.has_roles('Tenant Admin'): return redirect(prefixed_url_for('user_bp.tenant_consent', for_redirect=True)) else: return redirect(prefixed_url_for('user_bp.view_tenant_consents', tenant_id=tenant_id, for_redirect=True)) @user_bp.route('/consents///view', methods=['GET']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def view_consent_markdown(consent_type: str, version: str): """Render the consent markdown for a given type and version as an HTML fragment, using content_manager.""" try: current_app.logger.debug(f"Rendering markdown for {consent_type} version {version}") # Validate type against config valid_types = set(ConsentServices.get_required_consent_types()) if consent_type not in valid_types: for t in valid_types: if t.lower() == consent_type.lower(): consent_type = t break if consent_type not in valid_types: current_app.logger.warning(f"Unknown consent type requested for view: {consent_type}") return (render_template('user/partials/consent_markdown_fragment.html', markdown_content=f"Unknown consent type: {consent_type}"), 404) # Version must exist in ConsentVersion for the type cv = ConsentVersion.query.filter_by(consent_type=consent_type, consent_version=version).first() if not cv: current_app.logger.warning(f"Unknown consent version requested: type={consent_type}, version={version}") return (render_template('user/partials/consent_markdown_fragment.html', markdown_content=f"Document not found for version {version}"), 404) # Map consent type to content_manager content_type type_map = current_app.config.get('CONSENT_TYPE_MAP', {}) content_type = type_map.get(consent_type) if not content_type: current_app.logger.warning(f"No content_type mapping for consent type {consent_type}") return (render_template('user/partials/consent_markdown_fragment.html', markdown_content=f"Unknown content mapping for {consent_type}"), 404) # Parse major.minor and patch from version (e.g., 1.2.3 -> 1.2 and 1.2.3) major_minor, patch = VersionServices.split_version(version) # Use content_manager to read content content_data = content_manager.read_content(content_type, major_minor, patch) if not content_data or not content_data.get('content'): markdown_content = f"# Document not found\nThe consent document for {consent_type} version {version} could not be located." status = 404 else: markdown_content = content_data['content'] status = 200 return render_template('user/partials/consent_markdown_fragment.html', markdown_content=markdown_content), status except Exception as e: current_app.logger.error(f"Error in view_consent_markdown: {e}") return (render_template('user/partials/consent_markdown_fragment.html', markdown_content="Unexpected error rendering document."), 500) def reset_uniquifier(user): security.datastore.set_uniquifier(user) db.session.add(user) db.session.commit() send_reset_email(user) def get_notification_email(tenant_id, user_email=None): """ Determine which email address to use for notification. Priority: Provided email > Primary contact > Default email """ if user_email: return user_email # Try to find primary contact primary_contact = User.query.filter_by( tenant_id=tenant_id, is_primary_contact=True ).first() if primary_contact: return primary_contact.email return "pieter@askeveai.com" def send_api_key_notification(tenant_id, tenant_name, project_name, api_key, services, responsible_email=None): """ Send API key notification email """ recipient_email = get_notification_email(tenant_id, responsible_email) # Prepare email content context = { 'tenant_id': tenant_id, 'tenant_name': tenant_name, 'project_name': project_name, 'api_key': api_key, 'services': services, 'year': dt.now(tz.utc).year, 'promo_image_url': current_app.config.get('PROMOTIONAL_IMAGE_URL', 'https://static.askeveai.com/promo/default.jpg') } try: # Create email message msg = send_email( subject='Your new API-key from Ask Eve AI (Evie)', html=render_template('email/api_key_notification.html', **context), to_email=recipient_email, to_name=recipient_email, ) current_app.logger.info(f"API key notification sent to {recipient_email} for tenant {tenant_id}") return True except Exception as e: current_app.logger.error(f"Failed to send API key notification email: {str(e)}") return False @user_bp.route('/tenant_consents_history', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def tenant_consents_history(): tenant_id = session['tenant']['id'] config = get_tenant_consents_list_view(tenant_id) return render_list_view('list_view.html', **config) @user_bp.route('/handle_tenant_consents_history_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_tenant_consents_history_selection(): action = request.form.get('action') if action == 'view_consent_document': tenant_consent_identification = request.form.get('selected_row') if not tenant_consent_identification: flash('No consent selected', 'warning') return redirect(prefixed_url_for('user_bp.tenant_consents_history', for_redirect=True)) try: consent_id = ast.literal_eval(tenant_consent_identification).get('value') except Exception: flash('Invalid selection', 'danger') return redirect(prefixed_url_for('user_bp.tenant_consents_history', for_redirect=True)) tc = TenantConsent.query.get_or_404(consent_id) type_map = current_app.config.get('CONSENT_TYPE_MAP', {}) consent_type_dir = type_map.get(tc.consent_type) major_minor, patch = VersionServices.split_version(tc.consent_version) # Redirect to the fragment view; the template will render the fragment response as a full page if opened return redirect(prefixed_url_for( 'basic_bp.view_content', content_type=consent_type_dir, version=major_minor, patch=patch, for_redirect=True )) # Default: back to the history page return redirect(prefixed_url_for('user_bp.tenant_consents_history', for_redirect=True))