finance_billing.py 10.0 KB

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