378 lines
17 KiB
Python
378 lines
17 KiB
Python
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
|
|
from .list_views.entitlement_list_views import get_license_tiers_list_view, get_license_list_view
|
|
from .list_views.list_view_utils import render_list_view
|
|
|
|
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.license_tiers'))
|
|
else:
|
|
form_validation_failed(request, form)
|
|
|
|
return render_template('entitlements/license_tier.html', form=form)
|
|
|
|
|
|
@entitlements_bp.route('/license_tiers', methods=['GET', 'POST'])
|
|
@roles_accepted('Super User', 'Partner Admin')
|
|
def license_tiers():
|
|
config = get_license_tiers_list_view()
|
|
|
|
# Check if there was an error in getting the configuration
|
|
if config.get('error'):
|
|
return render_template("index.html")
|
|
|
|
return render_list_view('list_view.html', **config)
|
|
|
|
|
|
@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.license_tiers'))
|
|
|
|
|
|
@entitlements_bp.route('/license_tier/<int:license_tier_id>', 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/<int:license_tier_id>', 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/<int:license_id>', 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('/licenses')
|
|
@roles_accepted('Super User', 'Partner Admin', 'Tenant Admin')
|
|
def licenses():
|
|
config = get_license_list_view()
|
|
|
|
# Check if there was an error in getting the configuration
|
|
if config.get('error'):
|
|
return render_template("index.html")
|
|
|
|
return render_list_view('list_view.html', **config)
|
|
|
|
|
|
@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.licenses'))
|
|
|
|
|
|
@entitlements_bp.route('/license/<int:license_id>/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.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/<int:license_id>/periods/<int:period_id>/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('/view_licenses')
|
|
@roles_accepted('Super User', 'Partner Admin', 'Tenant Admin')
|
|
def view_licenses_redirect():
|
|
# Redirect to the new licenses route
|
|
return redirect(prefixed_url_for('entitlements_bp.licenses'))
|
|
|
|
|
|
@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.tenants'))
|
|
|
|
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)
|
|
|