from datetime import datetime as dt, timezone as tz, timedelta from flask import request, redirect, flash, render_template, Blueprint, session, current_app, jsonify from flask_security import roles_accepted, current_user from sqlalchemy.exc import SQLAlchemyError from sqlalchemy import or_, desc import ast from common.models.entitlements import License, LicenseTier, LicenseUsage, LicensePeriod, PeriodStatus from common.extensions import db, cache_manager from common.services.entitlements import LicenseTierServices from common.services.user import PartnerServices from common.services.user import UserServices from common.utils.eveai_exceptions import EveAIException from common.utils.security_utils import current_user_has_role from .entitlements_forms import LicenseTierForm, LicenseForm from common.utils.view_assistants import prepare_table_for_macro, form_validation_failed from common.utils.nginx_utils import prefixed_url_for from common.utils.document_utils import set_logging_information, update_logging_information entitlements_bp = Blueprint('entitlements_bp', __name__, url_prefix='/entitlements') @entitlements_bp.route('/license_tier', methods=['GET', 'POST']) @roles_accepted('Super User') def license_tier(): form = LicenseTierForm() if form.validate_on_submit(): current_app.logger.info("Adding License Tier") new_license_tier = LicenseTier() form.populate_obj(new_license_tier) set_logging_information(new_license_tier, dt.now(tz.utc)) try: db.session.add(new_license_tier) db.session.commit() except SQLAlchemyError as e: db.session.rollback() current_app.logger.error(f'Failed to add license tier to database. Error: {str(e)}') flash(f'Failed to add license tier to database. Error: {str(e)}', 'success') return render_template('entitlements/license_tier.html', form=form) current_app.logger.info(f"Successfully created license tier {new_license_tier.id}") flash(f"Successfully created tenant license tier {new_license_tier.id}", 'success') return redirect(prefixed_url_for('entitlements_bp.view_license_tiers')) else: form_validation_failed(request, form) return render_template('entitlements/license_tier.html', form=form) @entitlements_bp.route('/view_license_tiers', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin') def view_license_tiers(): page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) today = dt.now(tz.utc) query = LicenseTier.query.filter( or_( LicenseTier.end_date == None, LicenseTier.end_date >= today ) ) if current_user_has_role('Partner Admin'): try: license_tier_ids = PartnerServices.get_allowed_license_tier_ids() except EveAIException as e: flash(f"Cannot retrieve License Tiers: {str(e)}", 'danger') current_app.logger.error(f'Cannot retrieve License Tiers for partner: {str(e)}') return render_template("index.html") if license_tier_ids and len(license_tier_ids) > 0: query = query.filter(LicenseTier.id.in_(license_tier_ids)) query = query.order_by(LicenseTier.start_date.desc(), LicenseTier.id) pagination = query.paginate(page=page, per_page=per_page, error_out=False) license_tiers = pagination.items rows = prepare_table_for_macro(license_tiers, [('id', ''), ('name', ''), ('version', ''), ('start_date', ''), ('end_date', '')]) return render_template('entitlements/view_license_tiers.html', rows=rows, pagination=pagination, can_assign_license=UserServices.can_user_assign_license()) @entitlements_bp.route('/handle_license_tier_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin') def handle_license_tier_selection(): action = request.form['action'] if action == 'create_license_tier': return redirect(prefixed_url_for('entitlements_bp.license_tier')) license_tier_identification = request.form['selected_row'] license_tier_id = ast.literal_eval(license_tier_identification).get('value') match action: case 'edit_license_tier': return redirect(prefixed_url_for('entitlements_bp.edit_license_tier', license_tier_id=license_tier_id)) case 'create_license_for_tenant': return redirect(prefixed_url_for('entitlements_bp.create_license', license_tier_id=license_tier_id)) case 'associate_license_tier_to_partner': LicenseTierServices.associate_license_tier_with_partner(license_tier_id) # Add more conditions for other actions return redirect(prefixed_url_for('entitlements_bp.view_license_tiers')) @entitlements_bp.route('/license_tier/', methods=['GET', 'POST']) @roles_accepted('Super User') def edit_license_tier(license_tier_id): license_tier = LicenseTier.query.get_or_404(license_tier_id) # This will return a 404 if no license tier is found form = LicenseTierForm(obj=license_tier) if form.validate_on_submit(): # Populate the license_tier with form data form.populate_obj(license_tier) update_logging_information(license_tier, dt.now(tz.utc)) try: db.session.add(license_tier) db.session.commit() except SQLAlchemyError as e: db.session.rollback() current_app.logger.error(f'Failed to edit License Tier. Error: {str(e)}') flash(f'Failed to edit License Tier. Error: {str(e)}', 'danger') return render_template('entitlements/license_tier.html', form=form, license_tier_id=license_tier.id) flash('License Tier updated successfully.', 'success') return redirect( prefixed_url_for('entitlements_bp.edit_license_tier', license_tier_id=license_tier_id)) else: form_validation_failed(request, form) return render_template('entitlements/license_tier.html', form=form, license_tier_id=license_tier.id) @entitlements_bp.route('/create_license/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin') def create_license(license_tier_id): form = LicenseForm() tenant_id = session.get('tenant').get('id') currency = session.get('tenant').get('currency') readonly_fields = [] if current_user_has_role("Partner Admin"): # The Partner Admin can only set start & end dates, and allowed fields readonly_fields = [field.name for field in form if (field.name != 'nr_of_periods' and field.name != 'start_date' and not field.name.endswith('allowed'))] if request.method == 'GET': # Fetch the LicenseTier license_tier = LicenseTier.query.get_or_404(license_tier_id) # Prefill the form with LicenseTier data # Currency depending data if currency == '$': form.basic_fee.data = license_tier.basic_fee_d form.additional_storage_price.data = license_tier.additional_storage_price_d form.additional_embedding_price.data = license_tier.additional_embedding_price_d form.additional_interaction_token_price.data = license_tier.additional_interaction_token_price_d elif currency == '€': form.basic_fee.data = license_tier.basic_fee_e form.additional_storage_price.data = license_tier.additional_storage_price_e form.additional_embedding_price.data = license_tier.additional_embedding_price_e form.additional_interaction_token_price.data = license_tier.additional_interaction_token_price_e else: current_app.logger.error(f'Invalid currency {currency} for tenant {tenant_id} while creating license.') flash(f"Invalid currency {currency} for tenant {tenant_id} while creating license. " f"Check tenant's currency and try again.", 'danger') return redirect(prefixed_url_for('user_bp.edit_tenant', tenant_id=tenant_id)) # General data form.currency.data = currency form.max_storage_mb.data = license_tier.max_storage_mb form.additional_storage_bucket.data = license_tier.additional_storage_bucket form.included_embedding_mb.data = license_tier.included_embedding_mb form.additional_embedding_bucket.data = license_tier.additional_embedding_bucket form.included_interaction_tokens.data = license_tier.included_interaction_tokens form.additional_interaction_bucket.data = license_tier.additional_interaction_bucket form.overage_embedding.data = license_tier.standard_overage_embedding form.overage_interaction.data = license_tier.standard_overage_interaction else: # POST # Create a new License instance new_license = License( tenant_id=tenant_id, tier_id=license_tier_id, ) if form.validate_on_submit(): # Update the license with form data form.populate_obj(new_license) # Currency is added here again, as a form doesn't include disabled fields when passing it in the request new_license.currency = currency set_logging_information(new_license, dt.now(tz.utc)) try: db.session.add(new_license) db.session.commit() flash('License created successfully', 'success') return redirect(prefixed_url_for('entitlements_bp.edit_license', license_id=new_license.id)) except Exception as e: db.session.rollback() flash(f'Error creating license: {str(e)}', 'error') else: form_validation_failed(request, form) return render_template('entitlements/license.html', form=form, ext_readonly_fields=readonly_fields) @entitlements_bp.route('/license/', methods=['GET', 'POST']) @roles_accepted('Super User', 'Partner Admin') def edit_license(license_id): license = License.query.get_or_404(license_id) # This will return a 404 if no license tier is found form = LicenseForm(obj=license) readonly_fields = [] if len(license.periods) > 0: # There already are usage records linked to this license # Define which fields should be disabled readonly_fields = [field.name for field in form if field.name != 'nr_of_periods'] if current_user_has_role("Partner Admin"): # The Partner Admin can only set the nr_of_periods and allowed fields readonly_fields = [field.name for field in form if (field.name != 'nr_of_periods' and not field.name.endswith('allowed'))] cache_manager.license_cache.invalidate_tenant_license(license.tenant_id) if form.validate_on_submit(): # Populate the license with form data form.populate_obj(license) update_logging_information(license, dt.now(tz.utc)) try: db.session.add(license) db.session.commit() except SQLAlchemyError as e: db.session.rollback() current_app.logger.error(f'Failed to edit License. Error: {str(e)}') flash(f'Failed to edit License. Error: {str(e)}', 'danger') return render_template('entitlements/license.html', form=form) flash('License updated successfully.', 'success') return redirect( prefixed_url_for('entitlements_bp.edit_license', license_id=license_id)) else: form_validation_failed(request, form) return render_template('entitlements/license.html', form=form, ext_readonly_fields=readonly_fields) @entitlements_bp.route('/view_usages') @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def view_usages(): page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) tenant_id = session.get('tenant').get('id') query = LicenseUsage.query.filter_by(tenant_id=tenant_id).order_by(desc(LicenseUsage.id)) pagination = query.paginate(page=page, per_page=per_page) lus = pagination.items # prepare table data rows = prepare_table_for_macro(lus, [('id', ''), ('period_start_date', ''), ('period_end_date', ''), ('storage_mb_used', ''), ('embedding_mb_used', ''), ('interaction_total_tokens_used', '')]) # Render the users in a template return render_template('entitlements/view_usages.html', rows=rows, pagination=pagination) @entitlements_bp.route('/handle_usage_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_usage_selection(): usage_identification = request.form['selected_row'] usage_id = ast.literal_eval(usage_identification).get('value') the_usage = LicenseUsage.query.get_or_404(usage_id) action = request.form['action'] pass # Currently, no actions are defined @entitlements_bp.route('/view_licenses') @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def view_licenses(): page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) tenant_id = session.get('tenant').get('id') # Get current date in UTC current_date = dt.now(tz=tz.utc).date() # Query licenses for the tenant, with ordering and active status # TODO - Check validity query = ( License.query .join(LicenseTier) # Join with LicenseTier .filter(License.tenant_id == tenant_id) .add_columns( License.id, License.start_date, License.nr_of_periods, LicenseTier.name.label('license_tier_name'), # Access name through LicenseTier (License.start_date <= current_date).label('active') ) .order_by(License.start_date.desc()) ) pagination = query.paginate(page=page, per_page=per_page) lics = pagination.items # prepare table data rows = prepare_table_for_macro(lics, [('id', ''), ('license_tier_name', ''), ('start_date', ''), ('nr_of_periods', ''), ('active', '')]) # Render the licenses in a template return render_template('entitlements/view_licenses.html', rows=rows, pagination=pagination) @entitlements_bp.route('/handle_license_selection', methods=['POST']) @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def handle_license_selection(): license_identification = request.form['selected_row'] license_id = ast.literal_eval(license_identification).get('value') the_license = License.query.get_or_404(license_id) action = request.form['action'] match action: case 'edit_license': return redirect(prefixed_url_for('entitlements_bp.edit_license', license_id=license_id)) case 'view_periods': return redirect(prefixed_url_for('entitlements_bp.view_license_periods', license_id=license_id)) case _: return redirect(prefixed_url_for('entitlements_bp.view_licenses')) @entitlements_bp.route('/license//periods') @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def view_license_periods(license_id): license = License.query.get_or_404(license_id) # Verify user can access this license if not current_user.has_role('Super User'): tenant_id = session.get('tenant').get('id') if license.tenant_id != tenant_id: flash('Access denied to this license', 'danger') return redirect(prefixed_url_for('entitlements_bp.view_licenses')) # Get all periods for this license periods = (LicensePeriod.query .filter_by(license_id=license_id) .order_by(LicensePeriod.period_number) .all()) # Group related data for easy template access usage_by_period = {} payments_by_period = {} invoices_by_period = {} for period in periods: usage_by_period[period.id] = period.license_usage payments_by_period[period.id] = list(period.payments) invoices_by_period[period.id] = list(period.invoices) return render_template('entitlements/license_periods.html', license=license, periods=periods, usage_by_period=usage_by_period, payments_by_period=payments_by_period, invoices_by_period=invoices_by_period) @entitlements_bp.route('/license//periods//transition', methods=['POST']) @roles_accepted('Super User', 'Partner Admin') def transition_period_status(license_id, period_id): """Handle status transitions for license periods""" period = LicensePeriod.query.get_or_404(period_id) new_status = request.form.get('new_status') try: period.transition_status(PeriodStatus[new_status], current_user.id) db.session.commit() flash(f'Period {period.period_number} status updated to {new_status}', 'success') except ValueError as e: flash(f'Invalid status transition: {str(e)}', 'danger') except Exception as e: db.session.rollback() flash(f'Error updating status: {str(e)}', 'danger') return redirect(prefixed_url_for('entitlements_bp.view_license_periods', license_id=license_id)) @entitlements_bp.route('/active_usage') @roles_accepted('Super User', 'Partner Admin', 'Tenant Admin') def active_license_usage(): # Retrieve the active license period for the current tenant tenant_id = session.get('tenant', {}).get('id') if not tenant_id: flash('No active or pending license period found for this tenant', 'warning') return redirect(prefixed_url_for('user_bp.select_tenant')) active_period = LicensePeriod.query \ .join(License) \ .filter( License.tenant_id == tenant_id, LicensePeriod.status.in_([PeriodStatus.ACTIVE, PeriodStatus.PENDING]) ).first() if not active_period: flash('Geen actieve of pending licentieperiode gevonden voor deze tenant', 'warning') return render_template('entitlements/view_active_license_usage.html', active_period=None) # Bereken de percentages voor gebruik usage_data = {} if active_period.license_usage: # Storage percentage if active_period.max_storage_mb > 0: storage_used = active_period.license_usage.storage_mb_used or 0.0 usage_data['storage_used_rounded'] = round(storage_used, 2) usage_data['storage_percent'] = round(storage_used / active_period.max_storage_mb * 100, 2) else: usage_data['storage_percent'] = 0.0 # Embedding percentage if active_period.included_embedding_mb > 0: embedding_used = active_period.license_usage.embedding_mb_used or 0.0 usage_data['embedding_used_rounded'] = round(embedding_used, 2) usage_data['embedding_percent'] = round(embedding_used / active_period.included_embedding_mb * 100, 2) else: usage_data['embedding_percent'] = 0.0 # Interaction tokens percentage if active_period.included_interaction_tokens > 0: interaction_used = active_period.license_usage.interaction_total_tokens_used / 1_000_000 or 0.0 usage_data['interaction_used_rounded'] = round(interaction_used, 2) usage_data['interaction_percent'] = ( round(interaction_used / active_period.included_interaction_tokens * 100, 2)) else: usage_data['interaction_percent'] = 0 return render_template('entitlements/view_active_license_usage.html', active_period=active_period, usage_data=usage_data)