75484c0f73
API and persisted records use PowerOnModel system fields: - sysCreatedAt, sysCreatedBy, sysModifiedAt, sysModifiedBy Removed legacy JSON/DB field names: - _createdAt, _createdBy, _modifiedAt, _modifiedBy Frontend (frontend_nyla) and gateway call sites were updated accordingly. Database: - Bootstrap runs idempotent backfill (_migrateSystemFieldColumns) from old underscore columns and selected business duplicates into sys* where sys* IS NULL. - Re-run app bootstrap against each PostgreSQL database after deploy. - Optional: DROP INDEX IF EXISTS "idx_invitation_createdby" if an old index remains; new index: idx_invitation_syscreatedby on Invitation(sysCreatedBy). Tests: - RBAC integration tests aligned with current GROUP mandate filter and UserMandate-based UserConnection GROUP clause; buildRbacWhereClause(..., mandateId=...) must be passed explicitly (same as production request context).
1553 lines
61 KiB
Python
1553 lines
61 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Interface for Billing operations.
|
|
Manages billing accounts, transactions, and usage statistics.
|
|
|
|
All billing data is stored in the poweron_billing database.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, List, Optional, Union
|
|
from datetime import date, datetime, timedelta
|
|
import uuid
|
|
|
|
from modules.connectors.connectorDbPostgre import DatabaseConnector
|
|
from modules.shared.configuration import APP_CONFIG
|
|
from modules.shared.timeUtils import getUtcTimestamp
|
|
from modules.datamodels.datamodelUam import User, Mandate
|
|
from modules.datamodels.datamodelMembership import UserMandate
|
|
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResult
|
|
from modules.datamodels.datamodelBilling import (
|
|
BillingAccount,
|
|
BillingTransaction,
|
|
BillingSettings,
|
|
StripeWebhookEvent,
|
|
UsageStatistics,
|
|
BillingModelEnum,
|
|
AccountTypeEnum,
|
|
TransactionTypeEnum,
|
|
ReferenceTypeEnum,
|
|
PeriodTypeEnum,
|
|
BillingBalanceResponse,
|
|
BillingCheckResult,
|
|
parseBillingModelFromStoredValue,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _getAppDatabaseConnector() -> DatabaseConnector:
|
|
"""App DB connector (same config as UserMandate reads in this module)."""
|
|
return DatabaseConnector(
|
|
dbDatabase=APP_CONFIG.get("DB_DATABASE", "poweron_app"),
|
|
dbHost=APP_CONFIG.get("DB_HOST", "localhost"),
|
|
dbPort=int(APP_CONFIG.get("DB_PORT", "5432")),
|
|
dbUser=APP_CONFIG.get("DB_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET"),
|
|
)
|
|
|
|
|
|
def _getRootMandateIdFromAppDb(appDb: DatabaseConnector) -> Optional[str]:
|
|
"""Resolve root mandate id (name='root', isSystem=True) from app database."""
|
|
try:
|
|
rows = appDb.getRecordset(Mandate, recordFilter={"name": "root", "isSystem": True})
|
|
if rows:
|
|
rid = rows[0].get("id")
|
|
return str(rid) if rid is not None else None
|
|
except Exception as e:
|
|
logger.warning("Could not resolve root mandate id from app DB: %s", e)
|
|
return None
|
|
|
|
|
|
_cachedRootMandateId: Optional[str] = None
|
|
_rootMandateIdCacheResolved: bool = False
|
|
|
|
|
|
def _getCachedRootMandateId() -> Optional[str]:
|
|
"""Lazy-cached root mandate id (name=root, isSystem=True) for hot paths."""
|
|
global _cachedRootMandateId, _rootMandateIdCacheResolved
|
|
if not _rootMandateIdCacheResolved:
|
|
appDb = _getAppDatabaseConnector()
|
|
_cachedRootMandateId = _getRootMandateIdFromAppDb(appDb)
|
|
_rootMandateIdCacheResolved = True
|
|
return _cachedRootMandateId
|
|
|
|
|
|
# Singleton factory for BillingObjects instances
|
|
_billingInterfaces: Dict[str, "BillingObjects"] = {}
|
|
|
|
# Database name for billing
|
|
BILLING_DATABASE = "poweron_billing"
|
|
|
|
|
|
def getInterface(currentUser: User, mandateId: str = None) -> "BillingObjects":
|
|
"""
|
|
Factory function to get or create a BillingObjects instance.
|
|
|
|
Args:
|
|
currentUser: Current user object
|
|
mandateId: Mandate ID for context
|
|
|
|
Returns:
|
|
BillingObjects instance
|
|
"""
|
|
cacheKey = f"{currentUser.id}_{mandateId}"
|
|
|
|
if cacheKey not in _billingInterfaces:
|
|
_billingInterfaces[cacheKey] = BillingObjects(currentUser, mandateId)
|
|
else:
|
|
_billingInterfaces[cacheKey].setUserContext(currentUser, mandateId)
|
|
|
|
return _billingInterfaces[cacheKey]
|
|
|
|
|
|
def _getRootInterface() -> "BillingObjects":
|
|
"""Get interface with system access for bootstrap operations."""
|
|
from modules.security.rootAccess import getRootUser
|
|
rootUser = getRootUser()
|
|
return BillingObjects(rootUser, mandateId=None)
|
|
|
|
|
|
class BillingObjects:
|
|
"""
|
|
Interface for billing operations.
|
|
Manages accounts, transactions, settings, and statistics.
|
|
"""
|
|
|
|
def __init__(self, currentUser: Optional[User] = None, mandateId: str = None):
|
|
"""
|
|
Initialize the billing interface.
|
|
|
|
Args:
|
|
currentUser: Current user object
|
|
mandateId: Mandate ID for context
|
|
"""
|
|
self.currentUser = currentUser
|
|
self.userId = currentUser.id if currentUser else None
|
|
self.mandateId = mandateId
|
|
|
|
# Initialize database connection
|
|
self._initializeDatabase()
|
|
|
|
def _initializeDatabase(self):
|
|
"""Initialize database connection."""
|
|
self.db = DatabaseConnector(
|
|
dbDatabase=BILLING_DATABASE,
|
|
dbHost=APP_CONFIG.get('DB_HOST', 'localhost'),
|
|
dbPort=int(APP_CONFIG.get('DB_PORT', '5432')),
|
|
dbUser=APP_CONFIG.get('DB_USER'),
|
|
dbPassword=APP_CONFIG.get('DB_PASSWORD_SECRET')
|
|
)
|
|
|
|
def setUserContext(self, currentUser: User, mandateId: str = None):
|
|
"""
|
|
Update user context.
|
|
|
|
Args:
|
|
currentUser: Current user object
|
|
mandateId: Mandate ID for context
|
|
"""
|
|
self.currentUser = currentUser
|
|
self.userId = currentUser.id if currentUser else None
|
|
self.mandateId = mandateId
|
|
|
|
# =========================================================================
|
|
# BillingSettings Operations
|
|
# =========================================================================
|
|
|
|
def getSettings(self, mandateId: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get billing settings for a mandate.
|
|
|
|
Normalizes billingModel for API (legacy UNLIMITED → PREPAY_MANDATE) and persists once.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
|
|
Returns:
|
|
BillingSettings dict or None if not found
|
|
"""
|
|
try:
|
|
results = self.db.getRecordset(
|
|
BillingSettings,
|
|
recordFilter={"mandateId": mandateId}
|
|
)
|
|
if not results:
|
|
return None
|
|
row = dict(results[0])
|
|
raw_bm = row.get("billingModel")
|
|
parsed = parseBillingModelFromStoredValue(raw_bm)
|
|
if str(raw_bm or "").strip().upper() == "UNLIMITED":
|
|
try:
|
|
self.updateSettings(
|
|
row["id"],
|
|
{"billingModel": BillingModelEnum.PREPAY_MANDATE.value},
|
|
)
|
|
logger.info(
|
|
"Migrated billing settings for mandate %s: UNLIMITED → PREPAY_MANDATE",
|
|
mandateId,
|
|
)
|
|
except Exception as mig_err:
|
|
logger.warning(
|
|
"Could not persist billing model migration for mandate %s: %s",
|
|
mandateId,
|
|
mig_err,
|
|
)
|
|
row["billingModel"] = parsed.value
|
|
return row
|
|
except Exception as e:
|
|
logger.error(f"Error getting billing settings: {e}")
|
|
return None
|
|
|
|
def createSettings(self, settings: BillingSettings) -> Dict[str, Any]:
|
|
"""
|
|
Create billing settings for a mandate.
|
|
|
|
Args:
|
|
settings: BillingSettings object
|
|
|
|
Returns:
|
|
Created settings dict
|
|
"""
|
|
settingsDict = settings.model_dump(exclude_none=True)
|
|
return self.db.recordCreate(BillingSettings, settingsDict)
|
|
|
|
def updateSettings(self, settingsId: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Update billing settings.
|
|
|
|
Args:
|
|
settingsId: Settings ID
|
|
updates: Fields to update
|
|
|
|
Returns:
|
|
Updated settings dict or None
|
|
"""
|
|
return self.db.recordModify(BillingSettings, settingsId, updates)
|
|
|
|
def getOrCreateSettings(self, mandateId: str, defaultModel: BillingModelEnum = BillingModelEnum.PREPAY_MANDATE) -> Dict[str, Any]:
|
|
"""
|
|
Get or create billing settings for a mandate.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
defaultModel: Default billing model if creating
|
|
|
|
Returns:
|
|
BillingSettings dict
|
|
"""
|
|
existing = self.getSettings(mandateId)
|
|
if existing:
|
|
return existing
|
|
|
|
settings = BillingSettings(
|
|
mandateId=mandateId,
|
|
billingModel=defaultModel,
|
|
defaultUserCredit=0.0,
|
|
warningThresholdPercent=10.0,
|
|
notifyOnWarning=True,
|
|
)
|
|
return self.createSettings(settings)
|
|
|
|
# =========================================================================
|
|
# BillingAccount Operations
|
|
# =========================================================================
|
|
|
|
def getAccount(self, accountId: str) -> Optional[Dict[str, Any]]:
|
|
"""Get a billing account by ID."""
|
|
try:
|
|
results = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={"id": accountId}
|
|
)
|
|
return results[0] if results else None
|
|
except Exception as e:
|
|
logger.error(f"Error getting billing account: {e}")
|
|
return None
|
|
|
|
def getMandateAccount(self, mandateId: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get the mandate-level billing account.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
|
|
Returns:
|
|
BillingAccount dict or None
|
|
"""
|
|
try:
|
|
results = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={
|
|
"mandateId": mandateId,
|
|
"accountType": AccountTypeEnum.MANDATE.value
|
|
}
|
|
)
|
|
return results[0] if results else None
|
|
except Exception as e:
|
|
logger.error(f"Error getting mandate account: {e}")
|
|
return None
|
|
|
|
def getUserAccount(self, mandateId: str, userId: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get a user-level billing account within a mandate.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
userId: User ID
|
|
|
|
Returns:
|
|
BillingAccount dict or None
|
|
"""
|
|
try:
|
|
results = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={
|
|
"mandateId": mandateId,
|
|
"userId": userId,
|
|
"accountType": AccountTypeEnum.USER.value
|
|
}
|
|
)
|
|
return results[0] if results else None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user account: {e}")
|
|
return None
|
|
|
|
def getAccountsByMandate(self, mandateId: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all billing accounts for a mandate.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
|
|
Returns:
|
|
List of BillingAccount dicts
|
|
"""
|
|
try:
|
|
return self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={"mandateId": mandateId}
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error getting accounts for mandate: {e}")
|
|
return []
|
|
|
|
def createAccount(self, account: BillingAccount) -> Dict[str, Any]:
|
|
"""
|
|
Create a new billing account.
|
|
|
|
Args:
|
|
account: BillingAccount object
|
|
|
|
Returns:
|
|
Created account dict
|
|
"""
|
|
accountDict = account.model_dump(exclude_none=True)
|
|
return self.db.recordCreate(BillingAccount, accountDict)
|
|
|
|
def updateAccountBalance(self, accountId: str, newBalance: float) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Update account balance atomically.
|
|
|
|
Args:
|
|
accountId: Account ID
|
|
newBalance: New balance value
|
|
|
|
Returns:
|
|
Updated account dict or None
|
|
"""
|
|
return self.db.recordModify(BillingAccount, accountId, {"balance": newBalance})
|
|
|
|
def getOrCreateMandateAccount(self, mandateId: str, initialBalance: float = 0.0) -> Dict[str, Any]:
|
|
"""
|
|
Get or create a mandate-level billing account.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
initialBalance: Initial balance if creating
|
|
|
|
Returns:
|
|
BillingAccount dict
|
|
"""
|
|
existing = self.getMandateAccount(mandateId)
|
|
if existing:
|
|
return existing
|
|
|
|
account = BillingAccount(
|
|
mandateId=mandateId,
|
|
accountType=AccountTypeEnum.MANDATE,
|
|
balance=initialBalance,
|
|
enabled=True
|
|
)
|
|
return self.createAccount(account)
|
|
|
|
def getOrCreateUserAccount(self, mandateId: str, userId: str, initialBalance: float = 0.0) -> Dict[str, Any]:
|
|
"""
|
|
Get or create a user-level billing account.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
userId: User ID
|
|
initialBalance: Initial balance if creating
|
|
|
|
Returns:
|
|
BillingAccount dict
|
|
"""
|
|
existing = self.getUserAccount(mandateId, userId)
|
|
if existing:
|
|
return existing
|
|
|
|
account = BillingAccount(
|
|
mandateId=mandateId,
|
|
userId=userId,
|
|
accountType=AccountTypeEnum.USER,
|
|
balance=initialBalance,
|
|
enabled=True
|
|
)
|
|
created = self.createAccount(account)
|
|
|
|
# If initial balance > 0, create a SYSTEM credit transaction
|
|
if initialBalance > 0:
|
|
self.createTransaction(BillingTransaction(
|
|
accountId=created["id"],
|
|
transactionType=TransactionTypeEnum.CREDIT,
|
|
amount=initialBalance,
|
|
description="Initial credit for new user",
|
|
referenceType=ReferenceTypeEnum.SYSTEM
|
|
))
|
|
|
|
return created
|
|
|
|
def ensureAllMandateSettingsExist(self) -> int:
|
|
"""
|
|
Efficiently ensure all mandates have billing settings.
|
|
Creates default settings (PREPAY_MANDATE, 0 CHF) for mandates without settings.
|
|
Uses bulk queries to minimize database connections.
|
|
|
|
Returns:
|
|
Number of settings created
|
|
"""
|
|
try:
|
|
settingsCreated = 0
|
|
|
|
# Step 1: Get all existing billing settings in one query (from billing DB)
|
|
allSettings = self.db.getRecordset(BillingSettings)
|
|
existingMandateIds = set(s.get("mandateId") for s in allSettings if s.get("mandateId"))
|
|
|
|
# Step 2: Get all mandates from APP database (separate connection)
|
|
appDb = DatabaseConnector(
|
|
dbDatabase=APP_CONFIG.get('DB_DATABASE', 'poweron_app'),
|
|
dbHost=APP_CONFIG.get('DB_HOST', 'localhost'),
|
|
dbPort=int(APP_CONFIG.get('DB_PORT', '5432')),
|
|
dbUser=APP_CONFIG.get('DB_USER'),
|
|
dbPassword=APP_CONFIG.get('DB_PASSWORD_SECRET')
|
|
)
|
|
allMandates = appDb.getRecordset(Mandate, recordFilter={"enabled": True})
|
|
|
|
# Step 3: Create settings for mandates that don't have them
|
|
for mandate in allMandates:
|
|
mandateId = mandate.get("id")
|
|
if not mandateId or mandateId in existingMandateIds:
|
|
continue
|
|
|
|
# Create default billing settings
|
|
settings = BillingSettings(
|
|
mandateId=mandateId,
|
|
billingModel=BillingModelEnum.PREPAY_MANDATE,
|
|
defaultUserCredit=0.0,
|
|
warningThresholdPercent=10.0,
|
|
notifyOnWarning=True,
|
|
)
|
|
self.createSettings(settings)
|
|
existingMandateIds.add(mandateId) # Track newly created
|
|
settingsCreated += 1
|
|
|
|
if settingsCreated > 0:
|
|
logger.info(f"Created {settingsCreated} missing billing settings for mandates")
|
|
|
|
return settingsCreated
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error ensuring mandate settings exist: {e}")
|
|
return 0
|
|
|
|
def ensureAllUserAccountsExist(self) -> int:
|
|
"""
|
|
Ensure all users across all mandates have billing accounts.
|
|
User accounts are always created regardless of billing model (for audit trail).
|
|
Initial balance depends on billing model:
|
|
- PREPAY_USER: defaultUserCredit from settings only for the root mandate; other mandates get 0.0
|
|
- PREPAY_MANDATE: 0.0 (budget is on pool)
|
|
|
|
Uses bulk queries to minimize database connections.
|
|
|
|
Returns:
|
|
Number of accounts created
|
|
"""
|
|
try:
|
|
accountsCreated = 0
|
|
appDb = _getAppDatabaseConnector()
|
|
rootMandateId = _getCachedRootMandateId()
|
|
|
|
# Step 1: Get all billing settings (all mandates with settings get user accounts)
|
|
allSettings = self.db.getRecordset(BillingSettings)
|
|
billingMandates = {} # mandateId -> (billingModel, defaultCredit)
|
|
for s in allSettings:
|
|
billingModel = parseBillingModelFromStoredValue(s.get("billingModel")).value
|
|
mid = s.get("mandateId")
|
|
isRoot = rootMandateId is not None and str(mid) == str(rootMandateId)
|
|
if billingModel == BillingModelEnum.PREPAY_USER.value:
|
|
defaultCredit = (
|
|
float(s.get("defaultUserCredit", 0.0) or 0.0) if isRoot else 0.0
|
|
)
|
|
else:
|
|
defaultCredit = 0.0
|
|
billingMandates[mid] = (billingModel, defaultCredit)
|
|
|
|
if not billingMandates:
|
|
logger.debug("No billable mandates found, skipping account check")
|
|
return 0
|
|
|
|
# Step 2: Get all existing USER accounts in one query
|
|
allAccounts = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={"accountType": AccountTypeEnum.USER.value}
|
|
)
|
|
existingAccountKeys = set()
|
|
for acc in allAccounts:
|
|
key = (acc.get("mandateId"), acc.get("userId"))
|
|
existingAccountKeys.add(key)
|
|
|
|
# Step 3: Get all user-mandate combinations from APP database
|
|
allUserMandates = appDb.getRecordset(
|
|
UserMandate,
|
|
recordFilter={"enabled": True}
|
|
)
|
|
|
|
# Step 4: Create missing accounts
|
|
for um in allUserMandates:
|
|
mandateId = um.get("mandateId")
|
|
userId = um.get("userId")
|
|
|
|
if not mandateId or not userId:
|
|
continue
|
|
|
|
if mandateId not in billingMandates:
|
|
continue
|
|
|
|
key = (mandateId, userId)
|
|
if key in existingAccountKeys:
|
|
continue
|
|
|
|
billingModel, defaultCredit = billingMandates[mandateId]
|
|
|
|
account = BillingAccount(
|
|
mandateId=mandateId,
|
|
userId=userId,
|
|
accountType=AccountTypeEnum.USER,
|
|
balance=defaultCredit,
|
|
enabled=True
|
|
)
|
|
created = self.createAccount(account)
|
|
|
|
if defaultCredit > 0:
|
|
self.createTransaction(BillingTransaction(
|
|
accountId=created["id"],
|
|
transactionType=TransactionTypeEnum.CREDIT,
|
|
amount=defaultCredit,
|
|
description="Initial credit for new user",
|
|
referenceType=ReferenceTypeEnum.SYSTEM
|
|
))
|
|
|
|
existingAccountKeys.add(key)
|
|
accountsCreated += 1
|
|
|
|
if accountsCreated > 0:
|
|
logger.info(f"Created {accountsCreated} missing billing accounts")
|
|
|
|
return accountsCreated
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error ensuring user accounts exist: {e}")
|
|
return 0
|
|
|
|
# =========================================================================
|
|
# BillingTransaction Operations
|
|
# =========================================================================
|
|
|
|
def createTransaction(self, transaction: BillingTransaction, balanceAccountId: str = None) -> Dict[str, Any]:
|
|
"""
|
|
Create a new billing transaction and update account balance.
|
|
|
|
The transaction is always recorded against transaction.accountId (audit trail).
|
|
The balance is updated on balanceAccountId if provided, otherwise on transaction.accountId.
|
|
This allows recording a transaction on a user account (audit) while deducting
|
|
from a mandate pool account (shared budget).
|
|
|
|
Args:
|
|
transaction: BillingTransaction object
|
|
balanceAccountId: Optional account ID for balance update (defaults to transaction.accountId)
|
|
|
|
Returns:
|
|
Created transaction dict
|
|
"""
|
|
# Validate that the transaction's account exists
|
|
txAccount = self.getAccount(transaction.accountId)
|
|
if not txAccount:
|
|
raise ValueError(f"Transaction account {transaction.accountId} not found")
|
|
|
|
# Determine which account to update balance on
|
|
targetBalanceAccountId = balanceAccountId or transaction.accountId
|
|
if targetBalanceAccountId == transaction.accountId:
|
|
balanceAccount = txAccount
|
|
else:
|
|
balanceAccount = self.getAccount(targetBalanceAccountId)
|
|
if not balanceAccount:
|
|
raise ValueError(f"Balance account {targetBalanceAccountId} not found")
|
|
|
|
currentBalance = balanceAccount.get("balance", 0.0)
|
|
|
|
# Calculate new balance
|
|
if transaction.transactionType == TransactionTypeEnum.CREDIT:
|
|
newBalance = currentBalance + transaction.amount
|
|
elif transaction.transactionType == TransactionTypeEnum.DEBIT:
|
|
newBalance = currentBalance - transaction.amount
|
|
else: # ADJUSTMENT
|
|
newBalance = currentBalance + transaction.amount
|
|
|
|
# Create transaction record (always on transaction.accountId for audit)
|
|
transactionDict = transaction.model_dump(exclude_none=True)
|
|
created = self.db.recordCreate(BillingTransaction, transactionDict)
|
|
|
|
# Update balance on the target account
|
|
self.updateAccountBalance(targetBalanceAccountId, newBalance)
|
|
|
|
logger.info(f"Billing transaction created: {transaction.transactionType.value} {transaction.amount} CHF, "
|
|
f"audit={transaction.accountId}, balance on {targetBalanceAccountId}: {currentBalance} -> {newBalance}")
|
|
|
|
return created
|
|
|
|
def getTransactions(
|
|
self,
|
|
accountId: str,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
startDate: date = None,
|
|
endDate: date = None,
|
|
pagination: PaginationParams = None
|
|
) -> Union[List[Dict[str, Any]], PaginatedResult]:
|
|
"""
|
|
Get transactions for an account.
|
|
|
|
When pagination is provided, uses database-level pagination and returns
|
|
PaginatedResult. Otherwise falls back to in-memory filtering/sorting/slicing.
|
|
|
|
Args:
|
|
accountId: Account ID
|
|
limit: Maximum number of results (legacy path)
|
|
offset: Offset for pagination (legacy path)
|
|
startDate: Filter by start date (legacy path)
|
|
endDate: Filter by end date (legacy path)
|
|
pagination: PaginationParams for DB-level pagination
|
|
|
|
Returns:
|
|
PaginatedResult when pagination is provided, List of dicts otherwise
|
|
"""
|
|
try:
|
|
if pagination:
|
|
recordFilter = {"accountId": accountId}
|
|
result = self.db.getRecordsetPaginated(
|
|
BillingTransaction,
|
|
pagination=pagination,
|
|
recordFilter=recordFilter
|
|
)
|
|
return PaginatedResult(
|
|
items=result["items"],
|
|
totalItems=result["totalItems"],
|
|
totalPages=result["totalPages"]
|
|
)
|
|
|
|
filterDict = {"accountId": accountId}
|
|
results = self.db.getRecordset(BillingTransaction, recordFilter=filterDict)
|
|
|
|
if startDate or endDate:
|
|
filtered = []
|
|
for t in results:
|
|
createdAt = t.get("sysCreatedAt")
|
|
if createdAt:
|
|
tDate = createdAt.date() if isinstance(createdAt, datetime) else createdAt
|
|
if startDate and tDate < startDate:
|
|
continue
|
|
if endDate and tDate > endDate:
|
|
continue
|
|
filtered.append(t)
|
|
results = filtered
|
|
|
|
results.sort(key=lambda x: x.get("sysCreatedAt", ""), reverse=True)
|
|
|
|
return results[offset:offset + limit]
|
|
except Exception as e:
|
|
logger.error(f"Error getting transactions: {e}")
|
|
if pagination:
|
|
return PaginatedResult(items=[], totalItems=0, totalPages=0)
|
|
return []
|
|
|
|
def getTransactionsByMandate(
|
|
self,
|
|
mandateId: str,
|
|
limit: int = 100,
|
|
pagination: PaginationParams = None
|
|
) -> Union[List[Dict[str, Any]], PaginatedResult]:
|
|
"""
|
|
Get all transactions for a mandate (across all accounts).
|
|
|
|
When pagination is provided, collects all accountIds for the mandate and
|
|
issues a single DB query with SQL-level filtering, sorting, and pagination.
|
|
Otherwise falls back to per-account querying and in-memory merging.
|
|
|
|
Args:
|
|
mandateId: Mandate ID
|
|
limit: Maximum number of results (legacy path)
|
|
pagination: PaginationParams for DB-level pagination
|
|
|
|
Returns:
|
|
PaginatedResult when pagination is provided, List of dicts otherwise
|
|
"""
|
|
accounts = self.db.getRecordset(BillingAccount, recordFilter={"mandateId": mandateId})
|
|
accountIds = [acc["id"] for acc in accounts if acc.get("id")]
|
|
|
|
if not accountIds:
|
|
if pagination:
|
|
return PaginatedResult(items=[], totalItems=0, totalPages=0)
|
|
return []
|
|
|
|
if pagination:
|
|
result = self.db.getRecordsetPaginated(
|
|
BillingTransaction,
|
|
pagination=pagination,
|
|
recordFilter={"accountId": accountIds}
|
|
)
|
|
return PaginatedResult(
|
|
items=result["items"],
|
|
totalItems=result["totalItems"],
|
|
totalPages=result["totalPages"]
|
|
)
|
|
|
|
allTransactions = []
|
|
for account in accounts:
|
|
transactions = self.getTransactions(account["id"], limit=limit)
|
|
allTransactions.extend(transactions)
|
|
|
|
allTransactions.sort(key=lambda x: x.get("sysCreatedAt", ""), reverse=True)
|
|
return allTransactions[:limit]
|
|
|
|
# =========================================================================
|
|
# StripeWebhookEvent Operations (idempotency)
|
|
# =========================================================================
|
|
|
|
def getStripeWebhookEventByEventId(self, event_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Check if a Stripe event has already been processed (idempotency).
|
|
|
|
Args:
|
|
event_id: Stripe event ID (evt_xxx)
|
|
|
|
Returns:
|
|
Event record if exists, else None
|
|
"""
|
|
try:
|
|
results = self.db.getRecordset(
|
|
StripeWebhookEvent,
|
|
recordFilter={"event_id": event_id}
|
|
)
|
|
return results[0] if results else None
|
|
except Exception as e:
|
|
logger.error(f"Error checking Stripe webhook event: {e}")
|
|
return None
|
|
|
|
def createStripeWebhookEvent(self, event_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Record that a Stripe event has been processed.
|
|
|
|
Args:
|
|
event_id: Stripe event ID (evt_xxx)
|
|
|
|
Returns:
|
|
Created event record
|
|
"""
|
|
record = StripeWebhookEvent(event_id=event_id)
|
|
return self.db.recordCreate(StripeWebhookEvent, record.model_dump())
|
|
|
|
def getPaymentTransactionByReferenceId(self, referenceId: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Find an existing Stripe payment credit transaction by Checkout Session ID.
|
|
|
|
Args:
|
|
referenceId: Stripe Checkout Session ID (cs_xxx)
|
|
|
|
Returns:
|
|
Transaction record if found, else None
|
|
"""
|
|
try:
|
|
results = self.db.getRecordset(
|
|
BillingTransaction,
|
|
recordFilter={
|
|
"referenceType": ReferenceTypeEnum.PAYMENT.value,
|
|
"referenceId": referenceId,
|
|
}
|
|
)
|
|
return results[0] if results else None
|
|
except Exception as e:
|
|
logger.error(f"Error checking Stripe payment transaction by referenceId: {e}")
|
|
return None
|
|
|
|
# =========================================================================
|
|
# Balance Check Operations
|
|
# =========================================================================
|
|
|
|
def checkBalance(self, mandateId: str, userId: str, estimatedCost: float) -> BillingCheckResult:
|
|
"""
|
|
Check if there's sufficient balance for an operation.
|
|
|
|
- PREPAY_USER: user.balance >= estimatedCost
|
|
- PREPAY_MANDATE: mandate pool balance >= estimatedCost
|
|
|
|
User accounts are always ensured to exist (for audit trail).
|
|
Root mandate + PREPAY_USER: initial credit from settings.defaultUserCredit on first create.
|
|
Missing settings: treated as PREPAY_MANDATE with empty pool (strict).
|
|
"""
|
|
settings = self.getSettings(mandateId)
|
|
if not settings:
|
|
billingModel = BillingModelEnum.PREPAY_MANDATE
|
|
defaultCredit = 0.0
|
|
else:
|
|
billingModel = parseBillingModelFromStoredValue(settings.get("billingModel"))
|
|
defaultCredit = float(settings.get("defaultUserCredit", 0.0) or 0.0)
|
|
|
|
rootMandateId = _getCachedRootMandateId()
|
|
isRootMandate = rootMandateId is not None and str(mandateId) == str(rootMandateId)
|
|
if billingModel == BillingModelEnum.PREPAY_USER:
|
|
initialBalance = defaultCredit if isRootMandate else 0.0
|
|
else:
|
|
initialBalance = 0.0
|
|
self.getOrCreateUserAccount(mandateId, userId, initialBalance=initialBalance)
|
|
|
|
if billingModel == BillingModelEnum.PREPAY_USER:
|
|
account = self.getUserAccount(mandateId, userId)
|
|
currentBalance = account.get("balance", 0.0) if account else 0.0
|
|
else:
|
|
poolAccount = self.getOrCreateMandateAccount(mandateId)
|
|
currentBalance = poolAccount.get("balance", 0.0)
|
|
|
|
if currentBalance < estimatedCost:
|
|
return BillingCheckResult(
|
|
allowed=False,
|
|
reason="INSUFFICIENT_BALANCE",
|
|
currentBalance=currentBalance,
|
|
requiredAmount=estimatedCost,
|
|
billingModel=billingModel,
|
|
)
|
|
|
|
return BillingCheckResult(allowed=True, currentBalance=currentBalance, billingModel=billingModel)
|
|
|
|
def recordUsage(
|
|
self,
|
|
mandateId: str,
|
|
userId: str,
|
|
priceCHF: float,
|
|
workflowId: str = None,
|
|
featureInstanceId: str = None,
|
|
featureCode: str = None,
|
|
aicoreProvider: str = None,
|
|
aicoreModel: str = None,
|
|
description: str = "AI Usage",
|
|
processingTime: float = None,
|
|
bytesSent: int = None,
|
|
bytesReceived: int = None,
|
|
errorCount: int = None
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Record usage cost as a billing transaction.
|
|
|
|
Transaction is ALWAYS recorded on the user's account (clean audit trail).
|
|
Balance is deducted from the appropriate account based on billing model:
|
|
- PREPAY_USER: deduct from user's own balance
|
|
- PREPAY_MANDATE: deduct from mandate pool balance
|
|
"""
|
|
if priceCHF <= 0:
|
|
return None
|
|
|
|
settings = self.getSettings(mandateId)
|
|
if not settings:
|
|
logger.debug(f"No billing settings for mandate {mandateId}, skipping usage recording")
|
|
return None
|
|
|
|
billingModel = parseBillingModelFromStoredValue(settings.get("billingModel"))
|
|
|
|
# Transaction is ALWAYS on the user's account (audit trail)
|
|
userAccount = self.getOrCreateUserAccount(mandateId, userId)
|
|
|
|
transaction = BillingTransaction(
|
|
accountId=userAccount["id"],
|
|
transactionType=TransactionTypeEnum.DEBIT,
|
|
amount=priceCHF,
|
|
description=description,
|
|
referenceType=ReferenceTypeEnum.WORKFLOW,
|
|
workflowId=workflowId,
|
|
featureInstanceId=featureInstanceId,
|
|
featureCode=featureCode,
|
|
aicoreProvider=aicoreProvider,
|
|
aicoreModel=aicoreModel,
|
|
createdByUserId=userId,
|
|
processingTime=processingTime,
|
|
bytesSent=bytesSent,
|
|
bytesReceived=bytesReceived,
|
|
errorCount=errorCount
|
|
)
|
|
|
|
# Determine where to deduct balance
|
|
if billingModel == BillingModelEnum.PREPAY_USER:
|
|
return self.createTransaction(transaction)
|
|
if billingModel == BillingModelEnum.PREPAY_MANDATE:
|
|
poolAccount = self.getOrCreateMandateAccount(mandateId)
|
|
return self.createTransaction(transaction, balanceAccountId=poolAccount["id"])
|
|
return None
|
|
|
|
# =========================================================================
|
|
# Workflow Cost Query
|
|
# =========================================================================
|
|
|
|
def getWorkflowCost(self, workflowId: str) -> float:
|
|
"""Sum of all transaction amounts for a workflow."""
|
|
if not workflowId:
|
|
return 0.0
|
|
transactions = self.db.getRecordset(
|
|
BillingTransaction,
|
|
recordFilter={"workflowId": workflowId}
|
|
)
|
|
return sum(t.get("amount", 0.0) for t in transactions)
|
|
|
|
# =========================================================================
|
|
# Billing Model Switch Operations
|
|
# =========================================================================
|
|
|
|
def switchBillingModel(self, mandateId: str, oldModel: BillingModelEnum, newModel: BillingModelEnum) -> Dict[str, Any]:
|
|
"""
|
|
Switch billing model with budget migration logged as BillingTransactions.
|
|
|
|
PREPAY_MANDATE -> PREPAY_USER: pool debited, equal shares credited to user accounts.
|
|
PREPAY_USER -> PREPAY_MANDATE: user wallets debited, pool credited with sum.
|
|
"""
|
|
result = {"oldModel": oldModel.value, "newModel": newModel.value, "migratedAmount": 0.0, "userCount": 0}
|
|
|
|
if oldModel == newModel:
|
|
return result
|
|
|
|
if oldModel == BillingModelEnum.PREPAY_MANDATE and newModel == BillingModelEnum.PREPAY_USER:
|
|
poolAccount = self.getMandateAccount(mandateId)
|
|
userAccounts = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={"mandateId": mandateId, "accountType": AccountTypeEnum.USER.value}
|
|
)
|
|
poolBalance = poolAccount.get("balance", 0.0) if poolAccount else 0.0
|
|
n = len(userAccounts)
|
|
if poolAccount and poolBalance > 0:
|
|
self.createTransaction(
|
|
BillingTransaction(
|
|
accountId=poolAccount["id"],
|
|
transactionType=TransactionTypeEnum.DEBIT,
|
|
amount=poolBalance,
|
|
description="Model switch: distributed from mandate pool to user wallets",
|
|
referenceType=ReferenceTypeEnum.SYSTEM,
|
|
)
|
|
)
|
|
result["migratedAmount"] = poolBalance
|
|
if n > 0:
|
|
remaining = poolBalance
|
|
for i, acc in enumerate(userAccounts):
|
|
if i == n - 1:
|
|
share = round(remaining, 4)
|
|
else:
|
|
share = round(poolBalance / n, 4)
|
|
remaining -= share
|
|
if share > 0:
|
|
self.createTransaction(
|
|
BillingTransaction(
|
|
accountId=acc["id"],
|
|
transactionType=TransactionTypeEnum.CREDIT,
|
|
amount=share,
|
|
description="Model switch: share from mandate pool",
|
|
referenceType=ReferenceTypeEnum.SYSTEM,
|
|
)
|
|
)
|
|
result["userCount"] = n
|
|
logger.info(
|
|
"Switched %s MANDATE->USER: migrated %.4f CHF to %d user account(s) (transactions logged)",
|
|
mandateId,
|
|
result["migratedAmount"],
|
|
result["userCount"],
|
|
)
|
|
return result
|
|
|
|
if oldModel == BillingModelEnum.PREPAY_USER and newModel == BillingModelEnum.PREPAY_MANDATE:
|
|
userAccounts = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={"mandateId": mandateId, "accountType": AccountTypeEnum.USER.value}
|
|
)
|
|
totalUserBalance = sum(acc.get("balance", 0.0) for acc in userAccounts)
|
|
for acc in userAccounts:
|
|
b = acc.get("balance", 0.0)
|
|
if b > 0:
|
|
self.createTransaction(
|
|
BillingTransaction(
|
|
accountId=acc["id"],
|
|
transactionType=TransactionTypeEnum.DEBIT,
|
|
amount=b,
|
|
description="Model switch: consolidated to mandate pool",
|
|
referenceType=ReferenceTypeEnum.SYSTEM,
|
|
)
|
|
)
|
|
poolAccount = self.getOrCreateMandateAccount(mandateId, initialBalance=0.0)
|
|
if totalUserBalance > 0:
|
|
self.createTransaction(
|
|
BillingTransaction(
|
|
accountId=poolAccount["id"],
|
|
transactionType=TransactionTypeEnum.CREDIT,
|
|
amount=totalUserBalance,
|
|
description="Model switch: consolidated from user accounts",
|
|
referenceType=ReferenceTypeEnum.SYSTEM,
|
|
)
|
|
)
|
|
result["migratedAmount"] = totalUserBalance
|
|
result["userCount"] = len(userAccounts)
|
|
logger.info(
|
|
"Switched %s USER->MANDATE: consolidated %.4f CHF from %d users into pool (transactions logged)",
|
|
mandateId,
|
|
totalUserBalance,
|
|
len(userAccounts),
|
|
)
|
|
return result
|
|
|
|
if newModel == BillingModelEnum.PREPAY_MANDATE:
|
|
self.getOrCreateMandateAccount(mandateId, initialBalance=0.0)
|
|
|
|
return result
|
|
|
|
# =========================================================================
|
|
# Statistics Operations
|
|
# =========================================================================
|
|
|
|
def getUsageStatistics(
|
|
self,
|
|
accountId: str,
|
|
periodType: PeriodTypeEnum,
|
|
year: int,
|
|
month: int = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get usage statistics for an account.
|
|
|
|
Args:
|
|
accountId: Account ID
|
|
periodType: Period type (DAY, MONTH, YEAR)
|
|
year: Year
|
|
month: Month (for DAY period type)
|
|
|
|
Returns:
|
|
List of statistics dicts
|
|
"""
|
|
filterDict = {
|
|
"accountId": accountId,
|
|
"periodType": periodType.value
|
|
}
|
|
|
|
results = self.db.getRecordset(UsageStatistics, recordFilter=filterDict)
|
|
|
|
# Filter by year
|
|
filtered = [s for s in results if s.get("periodStart") and s["periodStart"].year == year]
|
|
|
|
# Filter by month if specified
|
|
if month and periodType == PeriodTypeEnum.DAY:
|
|
filtered = [s for s in filtered if s["periodStart"].month == month]
|
|
|
|
return sorted(filtered, key=lambda x: x.get("periodStart", date.min))
|
|
|
|
def calculateStatisticsFromTransactions(
|
|
self,
|
|
accountId: str,
|
|
startDate: date,
|
|
endDate: date
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Calculate statistics from transactions for a period.
|
|
|
|
Args:
|
|
accountId: Account ID
|
|
startDate: Start date
|
|
endDate: End date
|
|
|
|
Returns:
|
|
Statistics dict
|
|
"""
|
|
transactions = self.getTransactions(accountId, limit=10000, startDate=startDate, endDate=endDate)
|
|
|
|
# Filter only DEBIT transactions (usage)
|
|
debits = [t for t in transactions if t.get("transactionType") == TransactionTypeEnum.DEBIT.value]
|
|
|
|
totalCost = sum(t.get("amount", 0) for t in debits)
|
|
|
|
# Calculate by provider
|
|
costByProvider = {}
|
|
costByModel = {}
|
|
for t in debits:
|
|
provider = t.get("aicoreProvider", "unknown")
|
|
costByProvider[provider] = costByProvider.get(provider, 0) + t.get("amount", 0)
|
|
|
|
model = t.get("aicoreModel", "unknown")
|
|
costByModel[model] = costByModel.get(model, 0) + t.get("amount", 0)
|
|
|
|
# Calculate by feature
|
|
costByFeature = {}
|
|
for t in debits:
|
|
feature = t.get("featureCode", "unknown")
|
|
costByFeature[feature] = costByFeature.get(feature, 0) + t.get("amount", 0)
|
|
|
|
return {
|
|
"totalCostCHF": totalCost,
|
|
"transactionCount": len(debits),
|
|
"costByProvider": costByProvider,
|
|
"costByModel": costByModel,
|
|
"costByFeature": costByFeature
|
|
}
|
|
|
|
# =========================================================================
|
|
# Utility Methods
|
|
# =========================================================================
|
|
|
|
def getBalancesForUser(self, userId: str) -> List[BillingBalanceResponse]:
|
|
"""
|
|
Get all billing balances for a user across mandates.
|
|
|
|
Shows the effective available budget:
|
|
- PREPAY_USER: user's own account balance
|
|
- PREPAY_MANDATE: mandate pool balance (shared budget visible to user)
|
|
Args:
|
|
userId: User ID
|
|
|
|
Returns:
|
|
List of BillingBalanceResponse
|
|
"""
|
|
from modules.interfaces.interfaceDbApp import getRootInterface
|
|
|
|
balances = []
|
|
|
|
try:
|
|
# Use rootInterface (privileged, SysAdmin context) to bypass RBAC
|
|
# for mandate/user lookups. User access is verified via UserMandate membership.
|
|
rootInterface = getRootInterface()
|
|
userMandates = rootInterface.getUserMandates(userId)
|
|
|
|
for um in userMandates:
|
|
mandateId = getattr(um, 'mandateId', None) or (um.get("mandateId") if isinstance(um, dict) else None)
|
|
if not mandateId:
|
|
continue
|
|
|
|
mandate = rootInterface.getMandate(mandateId)
|
|
if not mandate:
|
|
continue
|
|
|
|
mandateName = getattr(mandate, 'label', None) or getattr(mandate, 'name', None) or (mandate.get("label") or mandate.get("name", "") if isinstance(mandate, dict) else "")
|
|
|
|
settings = self.getSettings(mandateId)
|
|
if not settings:
|
|
continue
|
|
|
|
billingModel = parseBillingModelFromStoredValue(settings.get("billingModel"))
|
|
|
|
if billingModel == BillingModelEnum.PREPAY_USER:
|
|
account = self.getOrCreateUserAccount(mandateId, userId)
|
|
if not account:
|
|
continue
|
|
balance = account.get("balance", 0.0)
|
|
warningThreshold = account.get("warningThreshold", 0.0)
|
|
elif billingModel == BillingModelEnum.PREPAY_MANDATE:
|
|
poolAccount = self.getOrCreateMandateAccount(mandateId)
|
|
if not poolAccount:
|
|
continue
|
|
balance = poolAccount.get("balance", 0.0)
|
|
warningThreshold = poolAccount.get("warningThreshold", 0.0)
|
|
else:
|
|
continue
|
|
|
|
balances.append(BillingBalanceResponse(
|
|
mandateId=mandateId,
|
|
mandateName=mandateName,
|
|
billingModel=billingModel,
|
|
balance=balance,
|
|
warningThreshold=warningThreshold,
|
|
isWarning=balance <= warningThreshold,
|
|
))
|
|
except Exception as e:
|
|
logger.error(f"Error getting balances for user: {e}")
|
|
|
|
return balances
|
|
|
|
def getTransactionsForUser(self, userId: str, limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all transactions for a user across all mandates they belong to.
|
|
Since transactions are always recorded on user accounts, we query
|
|
directly by user account - clean and simple.
|
|
|
|
Args:
|
|
userId: User ID
|
|
limit: Maximum number of results
|
|
|
|
Returns:
|
|
List of transaction dicts
|
|
"""
|
|
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
|
|
|
|
allTransactions = []
|
|
|
|
try:
|
|
appInterface = getAppInterface(self.currentUser)
|
|
userMandates = appInterface.getUserMandates(userId)
|
|
|
|
for um in userMandates:
|
|
mandateId = getattr(um, 'mandateId', None) or (um.get("mandateId") if isinstance(um, dict) else None)
|
|
if not mandateId:
|
|
continue
|
|
|
|
settings = self.getSettings(mandateId)
|
|
if not settings:
|
|
continue
|
|
|
|
# Get user's account in this mandate
|
|
userAccount = self.getUserAccount(mandateId, userId)
|
|
if not userAccount:
|
|
continue
|
|
|
|
# Get transactions for user's account (all transactions are on user accounts now)
|
|
transactions = self.getTransactions(userAccount["id"], limit=limit)
|
|
|
|
mandate = appInterface.getMandate(mandateId)
|
|
mandateName = ""
|
|
if mandate:
|
|
mandateName = getattr(mandate, 'label', None) or getattr(mandate, 'name', None) or (mandate.get("label") or mandate.get("name", "") if isinstance(mandate, dict) else "")
|
|
|
|
for t in transactions:
|
|
t["mandateId"] = mandateId
|
|
t["mandateName"] = mandateName
|
|
allTransactions.append(t)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting transactions for user: {e}")
|
|
|
|
allTransactions.sort(key=lambda x: x.get("sysCreatedAt", ""), reverse=True)
|
|
return allTransactions[:limit]
|
|
|
|
# =========================================================================
|
|
# Mandate View Operations (Admin-Level)
|
|
# =========================================================================
|
|
|
|
def getMandateBalances(self, mandateIds: List[str] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get mandate-level balances.
|
|
|
|
Args:
|
|
mandateIds: Optional list of mandate IDs to filter. If None, returns all.
|
|
|
|
Returns:
|
|
List of mandate balance dicts
|
|
"""
|
|
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
|
|
|
|
balances = []
|
|
|
|
try:
|
|
appInterface = getAppInterface(self.currentUser)
|
|
|
|
# Get settings for filtering
|
|
if mandateIds:
|
|
allSettings = [self.getSettings(mId) for mId in mandateIds]
|
|
allSettings = [s for s in allSettings if s]
|
|
else:
|
|
allSettings = self.db.getRecordset(BillingSettings)
|
|
|
|
for settings in allSettings:
|
|
mandateId = settings.get("mandateId")
|
|
if not mandateId:
|
|
continue
|
|
|
|
billingModel = parseBillingModelFromStoredValue(settings.get("billingModel"))
|
|
|
|
# Get mandate info
|
|
mandate = appInterface.getMandate(mandateId)
|
|
mandateName = ""
|
|
if mandate:
|
|
mandateName = getattr(mandate, 'label', None) or getattr(mandate, 'name', None) or (mandate.get("label") or mandate.get("name", "") if isinstance(mandate, dict) else "")
|
|
|
|
# Get user accounts count (always exist now for audit trail)
|
|
userAccounts = self.db.getRecordset(
|
|
BillingAccount,
|
|
recordFilter={"mandateId": mandateId, "accountType": AccountTypeEnum.USER.value}
|
|
)
|
|
userCount = len(userAccounts)
|
|
|
|
if billingModel == BillingModelEnum.PREPAY_USER:
|
|
totalBalance = sum(acc.get("balance", 0.0) for acc in userAccounts)
|
|
elif billingModel == BillingModelEnum.PREPAY_MANDATE:
|
|
poolAccount = self.getMandateAccount(mandateId)
|
|
totalBalance = poolAccount.get("balance", 0.0) if poolAccount else 0.0
|
|
else:
|
|
totalBalance = 0.0
|
|
|
|
balances.append({
|
|
"mandateId": mandateId,
|
|
"mandateName": mandateName,
|
|
"billingModel": billingModel.value,
|
|
"totalBalance": totalBalance,
|
|
"userCount": userCount,
|
|
"defaultUserCredit": float(settings.get("defaultUserCredit", 0.0) or 0.0),
|
|
"warningThresholdPercent": settings.get("warningThresholdPercent", 10.0),
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting mandate balances: {e}")
|
|
|
|
return balances
|
|
|
|
def getMandateTransactions(self, mandateIds: List[str] = None, limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all transactions for specified mandates.
|
|
|
|
Args:
|
|
mandateIds: Optional list of mandate IDs to filter. If None, returns all.
|
|
limit: Maximum number of results
|
|
|
|
Returns:
|
|
List of transaction dicts with mandate context
|
|
"""
|
|
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
|
|
|
|
allTransactions = []
|
|
|
|
try:
|
|
appInterface = getAppInterface(self.currentUser)
|
|
|
|
# Determine which mandates to query
|
|
if mandateIds:
|
|
targetMandateIds = mandateIds
|
|
else:
|
|
allSettings = self.db.getRecordset(BillingSettings)
|
|
targetMandateIds = [s.get("mandateId") for s in allSettings if s.get("mandateId")]
|
|
|
|
for mandateId in targetMandateIds:
|
|
transactions = self.getTransactionsByMandate(mandateId, limit=limit)
|
|
|
|
# Get mandate name
|
|
mandate = appInterface.getMandate(mandateId)
|
|
mandateName = ""
|
|
if mandate:
|
|
mandateName = getattr(mandate, 'label', None) or getattr(mandate, 'name', None) or (mandate.get("label") or mandate.get("name", "") if isinstance(mandate, dict) else "")
|
|
|
|
for t in transactions:
|
|
t["mandateId"] = mandateId
|
|
t["mandateName"] = mandateName
|
|
allTransactions.append(t)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting mandate transactions: {e}")
|
|
|
|
# Sort by creation date descending and limit
|
|
allTransactions.sort(key=lambda x: x.get("sysCreatedAt", ""), reverse=True)
|
|
return allTransactions[:limit]
|
|
|
|
# =========================================================================
|
|
# User View Operations (User-Level with RBAC)
|
|
# =========================================================================
|
|
|
|
def getUserBalancesForMandates(self, mandateIds: List[str] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all user-level balances for specified mandates.
|
|
|
|
Args:
|
|
mandateIds: Optional list of mandate IDs to filter. If None, returns all.
|
|
|
|
Returns:
|
|
List of user balance dicts with mandate and user context
|
|
"""
|
|
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
|
|
|
|
balances = []
|
|
|
|
try:
|
|
appInterface = getAppInterface(self.currentUser)
|
|
|
|
# Get all user accounts
|
|
accountFilter = {"accountType": AccountTypeEnum.USER.value}
|
|
allAccounts = self.db.getRecordset(BillingAccount, recordFilter=accountFilter)
|
|
|
|
# Filter by mandate if specified
|
|
if mandateIds:
|
|
allAccounts = [acc for acc in allAccounts if acc.get("mandateId") in mandateIds]
|
|
|
|
# Get all relevant settings in one query
|
|
settingsMap = {}
|
|
allSettings = self.db.getRecordset(BillingSettings)
|
|
for s in allSettings:
|
|
settingsMap[s.get("mandateId")] = s
|
|
|
|
# Get user info efficiently
|
|
userIds = list(set(acc.get("userId") for acc in allAccounts if acc.get("userId")))
|
|
userMap = {}
|
|
for userId in userIds:
|
|
user = appInterface.getUser(userId)
|
|
if user:
|
|
displayName = getattr(user, 'displayName', None) or (user.get("displayName") if isinstance(user, dict) else None)
|
|
username = getattr(user, 'username', None) or (user.get("username") if isinstance(user, dict) else None)
|
|
userMap[userId] = displayName or username or userId
|
|
|
|
# Get mandate info efficiently
|
|
mandateMap = {}
|
|
mandateIdList = list(set(acc.get("mandateId") for acc in allAccounts if acc.get("mandateId")))
|
|
for mandateId in mandateIdList:
|
|
mandate = appInterface.getMandate(mandateId)
|
|
if mandate:
|
|
mandateName = getattr(mandate, 'label', None) or getattr(mandate, 'name', None) or (mandate.get("label") or mandate.get("name", "") if isinstance(mandate, dict) else "")
|
|
mandateMap[mandateId] = mandateName
|
|
|
|
for account in allAccounts:
|
|
mandateId = account.get("mandateId")
|
|
userId = account.get("userId")
|
|
|
|
if not mandateId or not userId:
|
|
continue
|
|
|
|
settings = settingsMap.get(mandateId)
|
|
if not settings:
|
|
continue
|
|
|
|
balance = account.get("balance", 0.0)
|
|
warningThreshold = account.get("warningThreshold", 0.0)
|
|
|
|
balances.append({
|
|
"accountId": account.get("id"),
|
|
"mandateId": mandateId,
|
|
"mandateName": mandateMap.get(mandateId, ""),
|
|
"userId": userId,
|
|
"userName": userMap.get(userId, userId),
|
|
"balance": balance,
|
|
"warningThreshold": warningThreshold,
|
|
"isWarning": balance <= warningThreshold,
|
|
"enabled": account.get("enabled", True)
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting user balances for mandates: {e}")
|
|
|
|
return balances
|
|
|
|
def getUserTransactionsForMandates(self, mandateIds: List[str] = None, limit: int = 100) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all transactions for specified mandates.
|
|
All usage transactions are on user accounts (audit trail).
|
|
|
|
Args:
|
|
mandateIds: Optional list of mandate IDs to filter. If None, returns all.
|
|
limit: Maximum number of results
|
|
|
|
Returns:
|
|
List of transaction dicts with mandate and user context
|
|
"""
|
|
from modules.interfaces.interfaceDbApp import getInterface as getAppInterface
|
|
|
|
allTransactions = []
|
|
|
|
try:
|
|
appInterface = getAppInterface(self.currentUser)
|
|
|
|
# Get ALL accounts (both USER and MANDATE types) to cover all billing models
|
|
allAccounts = self.db.getRecordset(BillingAccount)
|
|
|
|
# Filter by mandate if specified
|
|
if mandateIds:
|
|
allAccounts = [acc for acc in allAccounts if acc.get("mandateId") in mandateIds]
|
|
|
|
# Build account to user/mandate mapping
|
|
accountMap = {}
|
|
for acc in allAccounts:
|
|
accountMap[acc.get("id")] = {
|
|
"mandateId": acc.get("mandateId"),
|
|
"userId": acc.get("userId")
|
|
}
|
|
|
|
# Get user info efficiently
|
|
userIds = list(set(acc.get("userId") for acc in allAccounts if acc.get("userId")))
|
|
userMap = {}
|
|
for userId in userIds:
|
|
user = appInterface.getUser(userId)
|
|
if user:
|
|
displayName = getattr(user, 'displayName', None) or (user.get("displayName") if isinstance(user, dict) else None)
|
|
username = getattr(user, 'username', None) or (user.get("username") if isinstance(user, dict) else None)
|
|
userMap[userId] = displayName or username or userId
|
|
|
|
# Get mandate info efficiently
|
|
mandateMap = {}
|
|
mandateIdList = list(set(acc.get("mandateId") for acc in allAccounts if acc.get("mandateId")))
|
|
for mandateId in mandateIdList:
|
|
mandate = appInterface.getMandate(mandateId)
|
|
if mandate:
|
|
mandateName = getattr(mandate, 'label', None) or getattr(mandate, 'name', None) or (mandate.get("label") or mandate.get("name", "") if isinstance(mandate, dict) else "")
|
|
mandateMap[mandateId] = mandateName
|
|
|
|
# Get transactions for all accounts and collect createdByUserIds
|
|
rawTransactions = []
|
|
for account in allAccounts:
|
|
accountId = account.get("id")
|
|
if not accountId:
|
|
continue
|
|
|
|
transactions = self.getTransactions(accountId, limit=limit)
|
|
accountInfo = accountMap.get(accountId, {})
|
|
mandateId = accountInfo.get("mandateId")
|
|
accountUserId = accountInfo.get("userId")
|
|
|
|
for t in transactions:
|
|
t["_accountUserId"] = accountUserId
|
|
t["_accountMandateId"] = mandateId
|
|
rawTransactions.append(t)
|
|
|
|
# Resolve createdByUserIds that are not yet in userMap
|
|
extraUserIds = set()
|
|
for t in rawTransactions:
|
|
cbUserId = t.get("createdByUserId")
|
|
if cbUserId and cbUserId not in userMap:
|
|
extraUserIds.add(cbUserId)
|
|
|
|
for uid in extraUserIds:
|
|
user = appInterface.getUser(uid)
|
|
if user:
|
|
displayName = getattr(user, 'displayName', None) or (user.get("displayName") if isinstance(user, dict) else None)
|
|
username = getattr(user, 'username', None) or (user.get("username") if isinstance(user, dict) else None)
|
|
userMap[uid] = displayName or username or uid
|
|
|
|
# Enrich transactions
|
|
for t in rawTransactions:
|
|
mandateId = t.pop("_accountMandateId", None)
|
|
accountUserId = t.pop("_accountUserId", None)
|
|
t["mandateId"] = mandateId
|
|
t["mandateName"] = mandateMap.get(mandateId, "")
|
|
# Prefer createdByUserId (per-transaction) over account-derived userId
|
|
txUserId = t.get("createdByUserId") or accountUserId
|
|
t["userId"] = txUserId
|
|
t["userName"] = userMap.get(txUserId, txUserId) if txUserId else None
|
|
allTransactions.append(t)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting user transactions for mandates: {e}")
|
|
|
|
# Sort by creation date descending and limit
|
|
allTransactions.sort(key=lambda x: x.get("sysCreatedAt", ""), reverse=True)
|
|
return allTransactions[:limit]
|