finance_billing.py 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213
  1. import logging
  2. from sqlalchemy import func, select
  3. from sqlalchemy.exc import IntegrityError, SQLAlchemyError
  4. from sqlalchemy.ext.asyncio import AsyncSession
  5. from backend.app.models.archive import PrintArchive
  6. from backend.app.models.finance import TransactionType, UserWallet, WalletTransaction
  7. from backend.app.services.finance_budget import is_billing_enabled, release_budget_reservation
  8. logger = logging.getLogger(__name__)
  9. async def _get_balance_after_for_transaction(
  10. db: AsyncSession,
  11. user_id: int,
  12. cost_center_id: int | None,
  13. amount: float,
  14. ) -> float:
  15. """Calculate balance_after for a transaction.
  16. For cost-center transactions: sum of ALL transactions for that cost center (global).
  17. For personal transactions (cost_center_id=None): user's wallet balance (personal).
  18. Args:
  19. user_id: The user making the transaction
  20. cost_center_id: The cost center (None for personal)
  21. amount: The transaction amount (positive/negative)
  22. Returns:
  23. The balance after this transaction would be applied
  24. """
  25. try:
  26. if cost_center_id is None:
  27. # Personal transaction: use user wallet balance
  28. wallet = (await db.execute(select(UserWallet).where(UserWallet.user_id == user_id))).scalar_one_or_none()
  29. if wallet is None:
  30. return float(amount)
  31. return float(wallet.balance) + amount
  32. else:
  33. # Cost-center transaction: sum of ALL transactions for this cost center (global, not per-user)
  34. result = await db.execute(
  35. select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)).where(
  36. WalletTransaction.cost_center_id == cost_center_id,
  37. )
  38. )
  39. current_balance = float(result.scalar() or 0.0)
  40. return current_balance + amount
  41. except SQLAlchemyError as e:
  42. logger.error(f"Database error in _get_balance_after_for_transaction: {e}", exc_info=True)
  43. raise
  44. def _calculate_partial_charge(
  45. archive: PrintArchive,
  46. base_cost: float,
  47. ) -> tuple[float, str]:
  48. """Calculate proportional charge for partial prints based on filament usage.
  49. Returns (charge_amount, description_suffix) where:
  50. - charge_amount: absolute cost to charge (0 if insufficient data)
  51. - description_suffix: reason/details for transaction description
  52. """
  53. try:
  54. # Only apply proportional calculation for non-completed prints
  55. if archive.status == "completed":
  56. return round(float(base_cost), 2), ""
  57. filament_used = float(archive.filament_used_grams or 0.0)
  58. filament_planned = None
  59. if archive.extra_data and isinstance(archive.extra_data, dict):
  60. filament_planned = archive.extra_data.get("filament_grams_total")
  61. if filament_planned is not None:
  62. filament_planned = float(filament_planned)
  63. # If we don't have reliable planned filament data, do not guess a partial charge.
  64. # Charging a failed/aborted print without an estimated baseline can overcharge users.
  65. if filament_planned is None or filament_planned <= 0:
  66. return 0.0, f"[{archive.status}: insufficient filament data]"
  67. # Calculate proportional cost
  68. filament_ratio = min(1.0, max(0.0, filament_used / filament_planned)) # Clamp to [0, 1]
  69. charge = float(base_cost) * filament_ratio
  70. # Round charges to 2 decimals for consistent persistence
  71. charge = round(charge, 2)
  72. suffix = f"[{archive.status}: {filament_ratio:.1%} filament ({filament_used:.1f}g/{filament_planned:.1f}g)]"
  73. return charge, suffix
  74. except ValueError as e:
  75. logger.error(f"Value error in _calculate_partial_charge: {e}", exc_info=True)
  76. raise
  77. async def apply_print_charge_for_archive(
  78. db: AsyncSession,
  79. archive_id: int,
  80. *,
  81. cost_center_id: int | None = None,
  82. print_run_id: str | None = None,
  83. ) -> bool:
  84. """Apply an idempotent wallet charge for a print archive.
  85. Charges completed prints at full cost, and partial/failed prints proportionally
  86. based on actual filament used vs. planned filament.
  87. Returns True when a new wallet transaction was created.
  88. """
  89. try:
  90. if not await is_billing_enabled(db):
  91. await release_budget_reservation(db, print_archive_id=archive_id, status="released")
  92. logger.info("Billing is disabled; skipping print charge for archive ID %s.", archive_id)
  93. return False
  94. archive = (
  95. await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id).with_for_update())
  96. ).scalar_one_or_none()
  97. if archive is None:
  98. logger.warning(f"Archive with ID {archive_id} not found.")
  99. return False
  100. if archive.wallet_charge_skipped:
  101. logger.info(f"Wallet charge skipped for archive ID {archive_id}.")
  102. return False
  103. # Accept completed, aborted, cancelled, and failed prints
  104. if archive.status not in ("completed", "aborted", "cancelled", "failed"):
  105. logger.info(f"Archive ID {archive_id} has status {archive.status}, which is not chargeable.")
  106. return False
  107. if archive.created_by_id is None:
  108. logger.warning(f"Archive ID {archive_id} has no creator ID.")
  109. return False
  110. base_cost = float(archive.cost or 0.0)
  111. if base_cost <= 0:
  112. logger.info(f"Base cost for archive ID {archive_id} is zero or negative.")
  113. return False
  114. tx_conditions = [WalletTransaction.transaction_type == TransactionType.PRINT_CHARGE.value]
  115. if print_run_id:
  116. tx_conditions.append(WalletTransaction.print_run_id == print_run_id)
  117. else:
  118. tx_conditions.append(WalletTransaction.print_archive_id == archive.id)
  119. existing_tx = (await db.execute(select(WalletTransaction).where(*tx_conditions))).scalar_one_or_none()
  120. if existing_tx is not None:
  121. logger.info(f"Transaction already exists for archive ID {archive_id}.")
  122. return False
  123. # Calculate charge (full for completed, partial for others)
  124. charge, reason_suffix = _calculate_partial_charge(archive, base_cost)
  125. if charge <= 0:
  126. await release_budget_reservation(db, print_archive_id=archive.id, status="released")
  127. logger.info(f"Calculated charge for archive ID {archive_id} is zero or negative.")
  128. return False
  129. actual_cost_center_id = cost_center_id if cost_center_id is not None else archive.cost_center_id
  130. wallet = (
  131. await db.execute(select(UserWallet).where(UserWallet.user_id == archive.created_by_id))
  132. ).scalar_one_or_none()
  133. if wallet is None:
  134. wallet = UserWallet(user_id=archive.created_by_id, balance=0.0, currency="EUR")
  135. db.add(wallet)
  136. await db.flush()
  137. logger.info(f"Created new wallet for user ID {archive.created_by_id}.")
  138. # Persist wallet balances rounded to cents
  139. new_wallet_balance = round(float(wallet.balance) - charge, 2)
  140. wallet.balance = new_wallet_balance
  141. label = archive.print_name or archive.filename or f"Archive {archive.id}"
  142. description = f"Print charge: {label}{' ' + reason_suffix if reason_suffix else ''}"
  143. balance_after = await _get_balance_after_for_transaction(
  144. db, archive.created_by_id, actual_cost_center_id, -charge
  145. )
  146. if balance_after is not None:
  147. balance_after = round(float(balance_after), 2)
  148. tx = WalletTransaction(
  149. user_id=archive.created_by_id,
  150. cost_center_id=actual_cost_center_id,
  151. transaction_type=TransactionType.PRINT_CHARGE.value,
  152. amount=-charge,
  153. balance_after=balance_after,
  154. description=description,
  155. created_by_user_id=None,
  156. print_run_id=print_run_id or archive.subtask_id,
  157. print_archive_id=archive.id,
  158. )
  159. db.add(tx)
  160. # Ensure the transaction is flushed to detect unique/index constraint violations
  161. try:
  162. await db.flush()
  163. except IntegrityError as e:
  164. # Another concurrent worker likely created the same transaction
  165. logger.info("Transaction already exists for archive ID %s (concurrent), skipping: %s", archive_id, e)
  166. await db.rollback()
  167. return False
  168. # Consume matching budget reservations after the transaction is persisted
  169. await release_budget_reservation(db, print_archive_id=archive.id, status="consumed")
  170. logger.info(f"Applied print charge for archive ID {archive_id}. New balance: {new_wallet_balance}.")
  171. return True
  172. except SQLAlchemyError as e:
  173. logger.error(f"Database error in apply_print_charge_for_archive: {e}", exc_info=True)
  174. return False
  175. except ValueError as e:
  176. logger.error(f"Value error in apply_print_charge_for_archive: {e}", exc_info=True)
  177. return False