401 lines
13 KiB
Python
401 lines
13 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Billing Service - Central service for billing operations.
|
|
|
|
Handles:
|
|
- Balance checks before AI operations
|
|
- Cost recording after AI operations
|
|
- Provider permission checks via RBAC
|
|
- Price calculation with markup
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
|
|
from modules.datamodels.datamodelUam import User
|
|
from modules.datamodels.datamodelBilling import (
|
|
BillingModelEnum,
|
|
BillingCheckResult,
|
|
TransactionTypeEnum,
|
|
ReferenceTypeEnum,
|
|
BillingTransaction,
|
|
BillingBalanceResponse,
|
|
)
|
|
from modules.interfaces.interfaceDbBilling import getInterface as getBillingInterface
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Markup percentage for internal pricing (50% = 1.5x)
|
|
BILLING_MARKUP_PERCENT = 50
|
|
|
|
# Singleton cache
|
|
_billingServices: Dict[str, "BillingService"] = {}
|
|
|
|
|
|
def getService(currentUser: User, mandateId: str, featureInstanceId: str = None, featureCode: str = None) -> "BillingService":
|
|
"""
|
|
Factory function to get or create a BillingService instance.
|
|
|
|
Args:
|
|
currentUser: Current user object
|
|
mandateId: Mandate ID for context
|
|
featureInstanceId: Optional feature instance ID
|
|
featureCode: Optional feature code (e.g., 'chatplayground', 'automation')
|
|
|
|
Returns:
|
|
BillingService instance
|
|
"""
|
|
cacheKey = f"{currentUser.id}_{mandateId}_{featureInstanceId}"
|
|
|
|
if cacheKey not in _billingServices:
|
|
_billingServices[cacheKey] = BillingService(currentUser, mandateId, featureInstanceId, featureCode)
|
|
else:
|
|
_billingServices[cacheKey].setContext(currentUser, mandateId, featureInstanceId, featureCode)
|
|
|
|
return _billingServices[cacheKey]
|
|
|
|
|
|
class BillingService:
|
|
"""
|
|
Central billing service for AI operations.
|
|
|
|
Responsibilities:
|
|
- Check balance before operations
|
|
- Record usage costs
|
|
- Apply pricing markup
|
|
- Check provider permissions via RBAC
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
currentUser: User,
|
|
mandateId: str,
|
|
featureInstanceId: str = None,
|
|
featureCode: str = None
|
|
):
|
|
"""
|
|
Initialize the billing service.
|
|
|
|
Args:
|
|
currentUser: Current user object
|
|
mandateId: Mandate ID
|
|
featureInstanceId: Optional feature instance ID
|
|
featureCode: Optional feature code
|
|
"""
|
|
self.currentUser = currentUser
|
|
self.mandateId = mandateId
|
|
self.featureInstanceId = featureInstanceId
|
|
self.featureCode = featureCode
|
|
|
|
# Get billing interface
|
|
self._billingInterface = getBillingInterface(currentUser, mandateId)
|
|
|
|
# Cache settings
|
|
self._settingsCache = None
|
|
|
|
def setContext(
|
|
self,
|
|
currentUser: User,
|
|
mandateId: str,
|
|
featureInstanceId: str = None,
|
|
featureCode: str = None
|
|
):
|
|
"""Update service context."""
|
|
self.currentUser = currentUser
|
|
self.mandateId = mandateId
|
|
self.featureInstanceId = featureInstanceId
|
|
self.featureCode = featureCode
|
|
self._billingInterface = getBillingInterface(currentUser, mandateId)
|
|
self._settingsCache = None
|
|
|
|
def _getSettings(self) -> Optional[Dict[str, Any]]:
|
|
"""Get billing settings with caching."""
|
|
if self._settingsCache is None:
|
|
self._settingsCache = self._billingInterface.getSettings(self.mandateId)
|
|
return self._settingsCache
|
|
|
|
# =========================================================================
|
|
# Price Calculation
|
|
# =========================================================================
|
|
|
|
def calculatePriceWithMarkup(self, basePriceCHF: float) -> float:
|
|
"""
|
|
Calculate final price with markup.
|
|
|
|
The AICore plugins return prices in their original currency (USD).
|
|
This method applies the configured markup percentage.
|
|
|
|
Args:
|
|
basePriceCHF: Base price from AI model (actually USD from provider)
|
|
|
|
Returns:
|
|
Final price in CHF with markup applied
|
|
"""
|
|
if basePriceCHF <= 0:
|
|
return 0.0
|
|
|
|
# Apply markup (50% = multiply by 1.5)
|
|
markup_multiplier = 1 + (BILLING_MARKUP_PERCENT / 100)
|
|
return round(basePriceCHF * markup_multiplier, 6)
|
|
|
|
# =========================================================================
|
|
# Balance Operations
|
|
# =========================================================================
|
|
|
|
def checkBalance(self, estimatedCost: float = 0.0) -> BillingCheckResult:
|
|
"""
|
|
Check if the current user/mandate has sufficient balance.
|
|
|
|
Args:
|
|
estimatedCost: Estimated cost of the operation (with markup applied)
|
|
|
|
Returns:
|
|
BillingCheckResult indicating if operation is allowed
|
|
"""
|
|
return self._billingInterface.checkBalance(
|
|
self.mandateId,
|
|
self.currentUser.id,
|
|
estimatedCost
|
|
)
|
|
|
|
def hasBalance(self, estimatedCost: float = 0.0) -> bool:
|
|
"""
|
|
Quick check if balance is sufficient.
|
|
|
|
Args:
|
|
estimatedCost: Estimated cost with markup
|
|
|
|
Returns:
|
|
True if operation is allowed
|
|
"""
|
|
result = self.checkBalance(estimatedCost)
|
|
return result.allowed
|
|
|
|
def getCurrentBalance(self) -> float:
|
|
"""
|
|
Get current balance for the user/mandate.
|
|
|
|
Returns:
|
|
Current balance in CHF
|
|
"""
|
|
result = self.checkBalance(0.0)
|
|
return result.currentBalance or 0.0
|
|
|
|
# =========================================================================
|
|
# Usage Recording
|
|
# =========================================================================
|
|
|
|
def recordUsage(
|
|
self,
|
|
priceCHF: float,
|
|
workflowId: str = None,
|
|
aicoreProvider: str = None,
|
|
description: str = None
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Record AI usage cost as a billing transaction.
|
|
|
|
This method:
|
|
1. Applies the pricing markup
|
|
2. Creates a DEBIT transaction
|
|
3. Updates the account balance
|
|
|
|
Args:
|
|
priceCHF: Base price from AI model (before markup)
|
|
workflowId: Optional workflow ID
|
|
aicoreProvider: AICore provider name (e.g., 'anthropic', 'openai')
|
|
description: Optional description
|
|
|
|
Returns:
|
|
Created transaction dict or None if not recorded
|
|
"""
|
|
if priceCHF <= 0:
|
|
return None
|
|
|
|
# Apply markup
|
|
finalPrice = self.calculatePriceWithMarkup(priceCHF)
|
|
|
|
if finalPrice <= 0:
|
|
return None
|
|
|
|
# Build description
|
|
if not description:
|
|
description = f"AI Usage: {aicoreProvider or 'unknown'}"
|
|
|
|
return self._billingInterface.recordUsage(
|
|
mandateId=self.mandateId,
|
|
userId=self.currentUser.id,
|
|
priceCHF=finalPrice,
|
|
workflowId=workflowId,
|
|
featureInstanceId=self.featureInstanceId,
|
|
featureCode=self.featureCode,
|
|
aicoreProvider=aicoreProvider,
|
|
description=description
|
|
)
|
|
|
|
# =========================================================================
|
|
# Provider Permission Check (via RBAC)
|
|
# =========================================================================
|
|
|
|
def isProviderAllowed(self, provider: str) -> bool:
|
|
"""
|
|
Check if the user has permission to use an AICore provider.
|
|
|
|
Uses RBAC to check for resource permission:
|
|
resource.aicore.{provider}
|
|
|
|
Args:
|
|
provider: Provider name (e.g., 'anthropic', 'openai')
|
|
|
|
Returns:
|
|
True if provider is allowed
|
|
"""
|
|
try:
|
|
from modules.security.rbac import RbacClass
|
|
from modules.datamodels.datamodelRbac import AccessRuleContext
|
|
from modules.security.rootAccess import getRootDbAppConnector
|
|
|
|
# Get database connector via established pattern
|
|
dbApp = getRootDbAppConnector()
|
|
|
|
rbac = RbacClass(dbApp, dbApp)
|
|
resourceKey = f"resource.aicore.{provider}"
|
|
|
|
# Check if user has view permission for this resource (view = use for RESOURCE context)
|
|
permissions = rbac.getUserPermissions(
|
|
self.currentUser,
|
|
AccessRuleContext.RESOURCE,
|
|
resourceKey,
|
|
mandateId=self.mandateId
|
|
)
|
|
|
|
return permissions.view
|
|
except Exception as e:
|
|
logger.warning(f"Error checking provider permission: {e}")
|
|
# Default to allowed if RBAC check fails
|
|
return True
|
|
|
|
def getallowedProviders(self) -> List[str]:
|
|
"""
|
|
Get list of AICore providers the user is allowed to use.
|
|
|
|
Returns:
|
|
List of allowed provider names
|
|
"""
|
|
try:
|
|
from modules.aicore.aicoreModelRegistry import modelRegistry
|
|
|
|
# Get all available providers
|
|
connectors = modelRegistry.discoverConnectors()
|
|
allProviders = [c.getConnectorType() for c in connectors]
|
|
|
|
# Filter by RBAC permissions
|
|
return [p for p in allProviders if self.isProviderAllowed(p)]
|
|
except Exception as e:
|
|
logger.warning(f"Error getting allowed providers: {e}")
|
|
return []
|
|
|
|
# =========================================================================
|
|
# Admin Operations
|
|
# =========================================================================
|
|
|
|
def addCredit(
|
|
self,
|
|
amount: float,
|
|
description: str = "Manual credit",
|
|
referenceType: ReferenceTypeEnum = ReferenceTypeEnum.ADMIN
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Add credit to the account (admin operation).
|
|
|
|
Args:
|
|
amount: Amount to credit (positive)
|
|
description: Transaction description
|
|
referenceType: Reference type (ADMIN, PAYMENT, SYSTEM)
|
|
|
|
Returns:
|
|
Created transaction dict or None
|
|
"""
|
|
if amount <= 0:
|
|
return None
|
|
|
|
settings = self._getSettings()
|
|
if not settings:
|
|
logger.warning(f"No billing settings for mandate {self.mandateId}")
|
|
return None
|
|
|
|
billingModel = BillingModelEnum(settings.get("billingModel", BillingModelEnum.UNLIMITED.value))
|
|
|
|
# Get or create account
|
|
if billingModel == BillingModelEnum.PREPAY_USER:
|
|
account = self._billingInterface.getOrCreateUserAccount(
|
|
self.mandateId,
|
|
self.currentUser.id,
|
|
initialBalance=0.0
|
|
)
|
|
else:
|
|
account = self._billingInterface.getOrCreateMandateAccount(
|
|
self.mandateId,
|
|
initialBalance=0.0
|
|
)
|
|
|
|
# Create credit transaction
|
|
transaction = BillingTransaction(
|
|
accountId=account["id"],
|
|
transactionType=TransactionTypeEnum.CREDIT,
|
|
amount=amount,
|
|
description=description,
|
|
referenceType=referenceType
|
|
)
|
|
|
|
return self._billingInterface.createTransaction(transaction)
|
|
|
|
# =========================================================================
|
|
# Statistics & Reporting
|
|
# =========================================================================
|
|
|
|
def getBalancesForUser(self) -> List[BillingBalanceResponse]:
|
|
"""
|
|
Get all billing balances for the current user.
|
|
|
|
Returns:
|
|
List of balance responses for each mandate
|
|
"""
|
|
return self._billingInterface.getBalancesForUser(self.currentUser.id)
|
|
|
|
def getTransactionHistory(self, limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get transaction history for the current mandate.
|
|
|
|
Args:
|
|
limit: Maximum number of transactions
|
|
|
|
Returns:
|
|
List of transactions
|
|
"""
|
|
return self._billingInterface.getTransactionsByMandate(self.mandateId, limit=limit)
|
|
|
|
|
|
# ============================================================================
|
|
# Exception Classes
|
|
# ============================================================================
|
|
|
|
class InsufficientBalanceException(Exception):
|
|
"""Raised when there's insufficient balance for an operation."""
|
|
|
|
def __init__(self, currentBalance: float, requiredAmount: float, message: str = None):
|
|
self.currentBalance = currentBalance
|
|
self.requiredAmount = requiredAmount
|
|
self.message = message or f"Insufficient balance. Current: {currentBalance:.2f} CHF, Required: {requiredAmount:.2f} CHF"
|
|
super().__init__(self.message)
|
|
|
|
|
|
class ProviderNotAllowedException(Exception):
|
|
"""Raised when a user doesn't have permission to use an AI provider."""
|
|
|
|
def __init__(self, provider: str, message: str = None):
|
|
self.provider = provider
|
|
self.message = message or f"Provider '{provider}' is not allowed for your role"
|
|
super().__init__(self.message)
|