diff --git a/backend/app/api/routes/archives.py b/backend/app/api/routes/archives.py index aefe9aecc..f4e8b2994 100644 --- a/backend/app/api/routes/archives.py +++ b/backend/app/api/routes/archives.py @@ -29,6 +29,7 @@ from backend.app.schemas.archive import ArchiveResponse, ArchiveSlim, ArchiveSta from backend.app.schemas.print_log import PrintLogResponse from backend.app.schemas.slicer import SliceRequest from backend.app.services.archive import ArchiveService +from backend.app.services.finance_budget import validate_print_budget from backend.app.utils.http import build_content_disposition from backend.app.utils.safe_path import safe_join_under from backend.app.utils.threemf_tools import ( @@ -4078,6 +4079,13 @@ async def reprint_archive( if not file_path.is_file(): raise HTTPException(404, "Archive file not found") + await validate_print_budget( + db, + cost_center_id=body.cost_center_id, + estimated_cost=body.estimated_cost, + current_user=user, + ) + plate_name = body.plate_name if not plate_name and body.plate_id is not None: plate_name = f"Plate {body.plate_id}" diff --git a/backend/app/api/routes/finance.py b/backend/app/api/routes/finance.py new file mode 100644 index 000000000..5f7fca1fb --- /dev/null +++ b/backend/app/api/routes/finance.py @@ -0,0 +1,1085 @@ +import calendar +from datetime import datetime, timezone +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import and_, case, func, or_, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_auth_if_enabled +from backend.app.core.database import get_db +from backend.app.core.permissions import Permission +from backend.app.models.archive import PrintArchive +from backend.app.models.finance import ( + BudgetReservation, + CostCenter, + CostCenterMember, + TransactionType, + UserWallet, + WalletTransaction, + normalize_transaction_type, +) +from backend.app.models.print_queue import PrintQueueItem +from backend.app.models.settings import Settings +from backend.app.models.user import User +from backend.app.schemas.finance import ( + CostCenterBudgetUpdateRequest, + CostCenterCreateRequest, + CostCenterDetailResponse, + CostCenterMemberRequest, + CostCenterMemberResponse, + CostCenterSummaryResponse, + CostCenterUpdateRequest, + ManualPrintRequest, + TransactionEditRequest, + WalletAdjustmentRequest, + WalletAdjustmentResponse, + WalletBalanceResponse, + WalletTransactionListResponse, + WalletTransactionResponse, +) + +router = APIRouter(prefix="/finance", tags=["finance"]) + + +def _serialize_wallet_transaction(tx: WalletTransaction) -> WalletTransactionResponse: + transaction_type = ( + tx.transaction_type.value if isinstance(tx.transaction_type, TransactionType) else tx.transaction_type + ) + return WalletTransactionResponse.model_construct( + id=tx.id, + user_id=tx.user_id, + cost_center_id=tx.cost_center_id, + transaction_type=transaction_type, + amount=tx.amount, + balance_after=tx.balance_after, + description=tx.description, + created_by_user_id=tx.created_by_user_id, + print_run_id=tx.print_run_id, + print_archive_id=tx.print_archive_id, + print_queue_id=tx.print_queue_id, + created_at=tx.created_at, + ) + + +def _clamp_day(year: int, month: int, desired_day: int) -> int: + return min(max(1, desired_day), calendar.monthrange(year, month)[1]) + + +async def _get_budget_window_start_utc(db: AsyncSession) -> datetime: + """Resolve monthly budget window start in UTC using configurable reset day/timezone. + + Defaults preserve current behavior: day=1, timezone=UTC. + """ + desired_day = 1 + tz_name = "UTC" + + result = await db.execute( + select(Settings).where(Settings.key.in_(["finance_budget_reset_day", "finance_budget_reset_timezone"])) + ) + for setting in result.scalars().all(): + if setting.key == "finance_budget_reset_day": + try: + parsed = int(setting.value) + if 1 <= parsed <= 31: + desired_day = parsed + except (TypeError, ValueError): + pass + elif setting.key == "finance_budget_reset_timezone": + value = (setting.value or "").strip() + if value: + tz_name = value + + try: + tz = ZoneInfo(tz_name) + except ZoneInfoNotFoundError: + tz = timezone.utc + + now_local = datetime.now(tz) + current_month_reset_day = _clamp_day(now_local.year, now_local.month, desired_day) + + if now_local.day >= current_month_reset_day: + start_local = datetime(now_local.year, now_local.month, current_month_reset_day, tzinfo=tz) + else: + prev_year = now_local.year + prev_month = now_local.month - 1 + if prev_month == 0: + prev_month = 12 + prev_year -= 1 + prev_month_reset_day = _clamp_day(prev_year, prev_month, desired_day) + start_local = datetime(prev_year, prev_month, prev_month_reset_day, tzinfo=tz) + + return start_local.astimezone(timezone.utc) + + +async def _get_cost_center_usage_maps( + db: AsyncSession, + cost_center_ids: list[int], +) -> tuple[dict[int, float], dict[int, float]]: + if not cost_center_ids: + return {}, {} + + spend_expr = case((WalletTransaction.amount < 0, -WalletTransaction.amount), else_=0.0) + + total_rows = await db.execute( + select(WalletTransaction.cost_center_id, func.coalesce(func.sum(spend_expr), 0.0)) + .where( + WalletTransaction.cost_center_id.in_(cost_center_ids), + WalletTransaction.cost_center_id.is_not(None), + ) + .group_by(WalletTransaction.cost_center_id) + ) + + budget_window_start_utc = await _get_budget_window_start_utc(db) + + month_rows = await db.execute( + select(WalletTransaction.cost_center_id, func.coalesce(func.sum(spend_expr), 0.0)) + .where( + WalletTransaction.cost_center_id.in_(cost_center_ids), + WalletTransaction.cost_center_id.is_not(None), + WalletTransaction.created_at >= budget_window_start_utc, + ) + .group_by(WalletTransaction.cost_center_id) + ) + + total_map = {int(center_id): float(value) for center_id, value in total_rows.all() if center_id is not None} + month_map = {int(center_id): float(value) for center_id, value in month_rows.all() if center_id is not None} + return total_map, month_map + + +async def _get_cost_center_balance_map( + db: AsyncSession, + cost_center_ids: list[int], +) -> dict[int, float]: + if not cost_center_ids: + return {} + + rows = await db.execute( + select(WalletTransaction.cost_center_id, func.coalesce(func.sum(WalletTransaction.amount), 0.0)) + .where( + WalletTransaction.cost_center_id.in_(cost_center_ids), + WalletTransaction.cost_center_id.is_not(None), + ) + .group_by(WalletTransaction.cost_center_id) + ) + return {int(center_id): float(value) for center_id, value in rows.all() if center_id is not None} + + +async def _get_cost_center_reserved_map( + db: AsyncSession, + cost_center_ids: list[int], +) -> dict[int, float]: + if not cost_center_ids: + return {} + + budget_rows = await db.execute( + select(BudgetReservation.cost_center_id, func.coalesce(func.sum(BudgetReservation.amount), 0.0)) + .where( + BudgetReservation.cost_center_id.in_(cost_center_ids), + BudgetReservation.status == "active", + ) + .group_by(BudgetReservation.cost_center_id) + ) + reserved_map = {int(center_id): float(value) for center_id, value in budget_rows.all() if center_id is not None} + + queue_rows = await db.execute( + select(PrintQueueItem.cost_center_id, func.coalesce(func.sum(PrintQueueItem.estimated_cost), 0.0)) + .where( + PrintQueueItem.cost_center_id.in_(cost_center_ids), + PrintQueueItem.status.in_(("pending", "printing")), + ) + .group_by(PrintQueueItem.cost_center_id) + ) + for center_id, value in queue_rows.all(): + if center_id is not None: + reserved_map[int(center_id)] = reserved_map.get(int(center_id), 0.0) + float(value or 0.0) + return reserved_map + + +def _budget_mode_and_limit(center: CostCenter) -> tuple[str, float | None]: + # Monthly takes precedence if legacy data still has both set. + if center.monthly_budget is not None: + return "monthly", float(center.monthly_budget) + if center.total_budget is not None: + return "total", float(center.total_budget) + return "none", None + + +def _to_cost_center_summary( + center: CostCenter, + *, + can_print: bool, + total_usage: float, + month_usage: float, + total_balance: float, + reserved: float = 0.0, +) -> CostCenterSummaryResponse: + budget_mode, budget_limit = _budget_mode_and_limit(center) + budget_used = month_usage if budget_mode == "monthly" else total_usage if budget_mode == "total" else None + budget_available = ( + max(0.0, budget_limit - budget_used - reserved) + if budget_limit is not None and budget_used is not None + else None + ) + + return CostCenterSummaryResponse( + id=center.id, + name=center.name, + is_private=center.is_private, + owner_user_id=center.owner_user_id, + is_active=center.is_active, + total_balance=total_balance, + total_budget=center.total_budget, + monthly_budget=center.monthly_budget, + budget_mode=budget_mode, + budget_limit=budget_limit, + budget_used=budget_used, + budget_available=budget_available, + can_print=can_print, + ) + + +async def _require_authenticated_user(current_user: User | None) -> User: + if current_user is None: + raise HTTPException(status_code=401, detail="Authentication required") + return current_user + + +def _has_cost_center_admin_access(user: User) -> bool: + return user.has_any_permission( + Permission.COST_CENTERS_READ_ALL.value, + Permission.COST_CENTERS_MODIFY.value, + Permission.COST_CENTERS_CREATE.value, + ) + + +async def _require_cost_center_admin_access(current_user: User | None) -> User: + user = await _require_authenticated_user(current_user) + if not _has_cost_center_admin_access(user): + raise HTTPException(status_code=403, detail="Missing required permissions for cost center administration") + return user + + +async def _get_or_create_wallet(db: AsyncSession, user_id: int) -> UserWallet: + result = await db.execute(select(UserWallet).where(UserWallet.user_id == user_id)) + wallet = result.scalar_one_or_none() + if wallet: + return wallet + + wallet = UserWallet(user_id=user_id, balance=0.0, currency="EUR") + db.add(wallet) + await db.flush() + await db.refresh(wallet) + return wallet + + +async def _get_user_or_404(db: AsyncSession, user_id: int) -> User: + result = await db.execute(select(User).where(User.id == user_id)) + user = result.scalar_one_or_none() + if user is None: + raise HTTPException(status_code=404, detail="User not found") + return user + + +async def _get_cost_center_or_404(db: AsyncSession, cost_center_id: int) -> CostCenter: + result = await db.execute( + select(CostCenter).options(selectinload(CostCenter.members)).where(CostCenter.id == cost_center_id) + ) + center = result.scalar_one_or_none() + if center is None: + raise HTTPException(status_code=404, detail="Cost center not found") + return center + + +def _to_balance_response(wallet: UserWallet) -> WalletBalanceResponse: + return WalletBalanceResponse( + user_id=wallet.user_id, + balance=wallet.balance, + currency=wallet.currency, + updated_at=wallet.updated_at, + ) + + +async def _build_personal_balance_map(db: AsyncSession, user_id: int) -> dict[int, float]: + result = await db.execute( + select( + WalletTransaction.id, + WalletTransaction.amount, + WalletTransaction.cost_center_id, + CostCenter.is_private, + CostCenter.owner_user_id, + ) + .where( + WalletTransaction.user_id == user_id, + ) + .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id) + .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) + ) + + running_balance = 0.0 + balance_map: dict[int, float] = {} + for transaction_id, amount, cost_center_id, is_private, owner_user_id in result.all(): + is_personal_cost_center = bool(cost_center_id is not None and is_private and owner_user_id == user_id) + if cost_center_id is None or is_personal_cost_center: + running_balance += float(amount) + balance_map[int(transaction_id)] = running_balance + + return balance_map + + +def _personal_balance_condition(user_id: int): + return or_( + WalletTransaction.cost_center_id.is_(None), + and_(CostCenter.is_private.is_(True), CostCenter.owner_user_id == user_id), + ) + + +async def _create_wallet_adjustment( + db: AsyncSession, + *, + target_user_id: int, + actor_user_id: int, + amount: float, + transaction_type: str, + description: str | None, + cost_center_id: int | None, +) -> WalletAdjustmentResponse: + transaction_type = normalize_transaction_type(transaction_type) + + if cost_center_id is not None: + await _get_cost_center_or_404(db, cost_center_id) + + wallet = await _get_or_create_wallet(db, target_user_id) + + # Calculate balance_after for this specific transaction context + if cost_center_id is None: + # Personal transaction: validate and update user wallet + new_balance = wallet.balance + amount + if new_balance < 0: + raise HTTPException(status_code=400, detail="Insufficient balance for withdrawal") + wallet.balance = new_balance + balance_after = new_balance + else: + # Cost-center transaction: validate against cost center balance only (global, not per-user) + result = await db.execute( + select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)).where( + WalletTransaction.cost_center_id == cost_center_id, + ) + ) + current_cc_balance = float(result.scalar() or 0.0) + new_cc_balance = current_cc_balance + amount + if new_cc_balance < 0: + raise HTTPException(status_code=400, detail="Insufficient cost center balance for withdrawal") + balance_after = new_cc_balance + # Do NOT update wallet.balance for cost-center transactions + + tx = WalletTransaction( + user_id=target_user_id, + cost_center_id=cost_center_id, + transaction_type=transaction_type, + amount=amount, + balance_after=balance_after, + description=description, + created_by_user_id=actor_user_id, + ) + db.add(tx) + await db.flush() + await db.commit() + await db.refresh(wallet) + await db.refresh(tx) + + # Return appropriate balance based on transaction type + if cost_center_id is None: + # Personal transaction: return user wallet balance + response_balance = _to_balance_response(wallet) + else: + # Cost-center transaction: return cost-center balance as if it were a wallet + response_balance = WalletBalanceResponse( + user_id=target_user_id, + balance=balance_after, + currency=wallet.currency, + updated_at=tx.created_at, + ) + + return WalletAdjustmentResponse( + transaction=_serialize_wallet_transaction(tx), + balance=response_balance, + ) + + +@router.get("/me/balance", response_model=WalletBalanceResponse) +async def get_my_balance( + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_OWN), +): + """Return the current user's wallet balance.""" + user = await _require_authenticated_user(current_user) + wallet = await _get_or_create_wallet(db, user.id) + personal_balance_result = await db.execute( + select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)) + .select_from(WalletTransaction) + .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id) + .where(WalletTransaction.user_id == user.id, _personal_balance_condition(user.id)) + ) + personal_balance = float(personal_balance_result.scalar_one() or 0.0) + return WalletBalanceResponse( + user_id=user.id, + balance=personal_balance, + currency=wallet.currency, + updated_at=wallet.updated_at, + ) + + +@router.get("/me/transactions", response_model=WalletTransactionListResponse) +async def get_my_transactions( + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_OWN), +): + """Return wallet ledger entries for the current user.""" + user = await _require_authenticated_user(current_user) + + total_result = await db.execute( + select(func.count(WalletTransaction.id)).where(WalletTransaction.user_id == user.id) + ) + total = int(total_result.scalar_one() or 0) + + result = await db.execute( + select(WalletTransaction) + .where(WalletTransaction.user_id == user.id) + .order_by(WalletTransaction.created_at.desc(), WalletTransaction.id.desc()) + .limit(limit) + .offset(offset) + ) + transactions = result.scalars().all() + personal_balance_map = await _build_personal_balance_map(db, user.id) + return WalletTransactionListResponse( + items=[ + _serialize_wallet_transaction(tx).model_copy( + update={"balance_after": personal_balance_map.get(tx.id, tx.balance_after)} + ) + for tx in transactions + ], + total=total, + limit=limit, + offset=offset, + ) + + +@router.get("/transactions", response_model=WalletTransactionListResponse) +async def get_all_transactions( + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), + user_id: int | None = Query(None, description="Optional filter by user id"), + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL), +): + """Return wallet ledger entries across users for admin finance view.""" + await _require_authenticated_user(current_user) + + conditions = [] + if user_id is not None: + await _get_user_or_404(db, user_id) + conditions.append(WalletTransaction.user_id == user_id) + + total_result = await db.execute(select(func.count(WalletTransaction.id)).where(*conditions)) + total = int(total_result.scalar_one() or 0) + + result = await db.execute( + select(WalletTransaction) + .where(*conditions) + .order_by(WalletTransaction.created_at.desc(), WalletTransaction.id.desc()) + .limit(limit) + .offset(offset) + ) + transactions = result.scalars().all() + return WalletTransactionListResponse( + items=[_serialize_wallet_transaction(tx) for tx in transactions], + total=total, + limit=limit, + offset=offset, + ) + + +async def _rebuild_wallet_ledger_for_user(db: AsyncSession, user_id: int) -> None: + """Recompute `balance_after` for all wallet transactions of a user. + + - Personal transactions (cost_center_id=None): running balance per user + - Cost-center transactions: running balance GLOBAL for entire cost center (not per-user) + - Also updates the user's wallet balance (sum of personal transactions only) + """ + result = await db.execute( + select(WalletTransaction) + .where(WalletTransaction.user_id == user_id) + .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) + ) + user_transactions = result.scalars().all() + + # Handle personal transactions (cost_center_id=None) + personal_balance = 0.0 + for tx in user_transactions: + if tx.cost_center_id is None: + personal_balance += float(tx.amount) + tx.balance_after = personal_balance + db.add(tx) + + for cc_id in {tx.cost_center_id for tx in user_transactions if tx.cost_center_id is not None}: + # Get all transactions for this cost center (all users, all time) + result_all_cc = await db.execute( + select(WalletTransaction) + .where(WalletTransaction.cost_center_id == cc_id) + .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) + ) + all_cc_transactions = result_all_cc.scalars().all() + + running = 0.0 + for tx in all_cc_transactions: + running += float(tx.amount) + tx.balance_after = running + db.add(tx) + + # Update user wallet balance (sum of all personal transactions only) + wallet = await _get_or_create_wallet(db, user_id) + wallet.balance = personal_balance + await db.flush() + await db.commit() + + +@router.delete("/transactions/{transaction_id}") +async def delete_transaction( + transaction_id: int, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL), +): + """Delete a wallet transaction and rebuild the user's ledger to keep balances consistent.""" + await _require_authenticated_user(current_user) + + result = await db.execute(select(WalletTransaction).where(WalletTransaction.id == transaction_id)) + tx = result.scalar_one_or_none() + if tx is None: + raise HTTPException(status_code=404, detail="Transaction not found") + + user_id = tx.user_id + + if tx.transaction_type == "print_charge" and tx.print_archive_id is not None: + archive_result = await db.execute(select(PrintArchive).where(PrintArchive.id == tx.print_archive_id)) + archive = archive_result.scalar_one_or_none() + if archive is not None: + archive.wallet_charge_skipped = True + db.add(archive) + + await db.delete(tx) + await db.flush() + + await _rebuild_wallet_ledger_for_user(db, user_id) + + return {"status": "success"} + + +@router.patch("/transactions/{transaction_id}", response_model=WalletTransactionResponse) +async def edit_transaction( + transaction_id: int, + request: TransactionEditRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL), +): + """Edit a wallet transaction (user_id, cost_center_id, amount, description) and rebuild ledger.""" + await _require_authenticated_user(current_user) + + result = await db.execute(select(WalletTransaction).where(WalletTransaction.id == transaction_id)) + tx = result.scalar_one_or_none() + if tx is None: + raise HTTPException(status_code=404, detail="Transaction not found") + + # Apply edits + if request.user_id is not None: + tx.user_id = request.user_id + + if request.cost_center_id is not None: + tx.cost_center_id = request.cost_center_id + + if request.amount is not None: + tx.amount = request.amount + + if request.description is not None: + # Append "(Admin edit)" marker if not already present + new_desc = request.description + if not new_desc.endswith("(Admin edit)"): + new_desc = f"{new_desc} (Admin edit)" + tx.description = new_desc + + db.add(tx) + await db.flush() + + # Rebuild full ledger using the current session + from backend.app.core.database import repair_wallet_ledger_internal + + await repair_wallet_ledger_internal(db) + + return tx + + +@router.post("/transactions/manual", response_model=WalletTransactionResponse) +async def create_manual_print( + request: ManualPrintRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL), +): + """Create a manual print charge transaction (for admin purposes).""" + await _require_authenticated_user(current_user) + + from datetime import timezone + + # Use provided created_at or current time + created_at = request.created_at or datetime.now(timezone.utc) + + # Ensure manual print charges are negative amounts (charges reduce wallet) + amount = request.amount + if amount > 0: + amount = -abs(amount) + + # Create transaction + tx = WalletTransaction( + user_id=request.user_id, + cost_center_id=request.cost_center_id, + transaction_type=TransactionType.MANUAL_ADJUSTMENT.value, + amount=amount, + balance_after=None, # Will be set by repair_wallet_ledger_internal + description=request.description or "Manual print charge", + created_by_user_id=current_user.id if current_user else None, + created_at=created_at, + ) + db.add(tx) + await db.flush() + + # Rebuild full ledger using the current session + from backend.app.core.database import repair_wallet_ledger_internal + + await repair_wallet_ledger_internal(db) + + return tx + + +@router.get("/cost-centers/mine", response_model=list[CostCenterSummaryResponse]) +async def get_my_cost_centers( + db: AsyncSession = Depends(get_db), + current_user: User | None = Depends(require_auth_if_enabled), +): + """Return private and assigned cost centers for the current user.""" + user = await _require_authenticated_user(current_user) + + result = await db.execute( + select(CostCenter, CostCenterMember.can_print) + .outerjoin( + CostCenterMember, + (CostCenterMember.cost_center_id == CostCenter.id) & (CostCenterMember.user_id == user.id), + ) + .where( + CostCenter.is_active.is_(True), + or_( + (CostCenter.is_private.is_(True) & (CostCenter.owner_user_id == user.id)), + (CostCenterMember.user_id == user.id), + ), + ) + .order_by(CostCenter.is_private.desc(), CostCenter.name.asc()) + ) + + rows = result.all() + centers_only = [center for center, _ in rows] + center_ids = [center.id for center in centers_only] + total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, center_ids) + total_balance_map = await _get_cost_center_balance_map(db, center_ids) + reserved_map = await _get_cost_center_reserved_map(db, center_ids) + + centers: list[CostCenterSummaryResponse] = [] + for center, can_print in rows: + centers.append( + _to_cost_center_summary( + center, + can_print=True if center.is_private and center.owner_user_id == user.id else bool(can_print), + total_usage=total_usage_map.get(center.id, 0.0), + month_usage=month_usage_map.get(center.id, 0.0), + total_balance=total_balance_map.get(center.id, 0.0), + reserved=reserved_map.get(center.id, 0.0), + ) + ) + + return centers + + +@router.get("/users/{user_id}/balance", response_model=WalletBalanceResponse) +async def get_user_balance( + user_id: int, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL), +): + """Return a specific user's wallet balance.""" + await _require_authenticated_user(current_user) + user = await _get_user_or_404(db, user_id) + wallet = await _get_or_create_wallet(db, user.id) + return _to_balance_response(wallet) + + +@router.get("/users/{user_id}/transactions", response_model=list[WalletTransactionResponse]) +async def get_user_transactions( + user_id: int, + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL), +): + """Return wallet ledger entries for a specific user.""" + await _require_authenticated_user(current_user) + await _get_user_or_404(db, user_id) + + result = await db.execute( + select(WalletTransaction) + .where(WalletTransaction.user_id == user_id) + .order_by(WalletTransaction.created_at.desc(), WalletTransaction.id.desc()) + .limit(limit) + .offset(offset) + ) + return [_serialize_wallet_transaction(tx) for tx in result.scalars().all()] + + +@router.post("/users/{user_id}/deposit", response_model=WalletAdjustmentResponse) +async def deposit_user_balance( + user_id: int, + body: WalletAdjustmentRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Add funds to a user's wallet.""" + actor = await _require_authenticated_user(current_user) + await _get_user_or_404(db, user_id) + return await _create_wallet_adjustment( + db, + target_user_id=user_id, + actor_user_id=actor.id, + amount=body.amount, + transaction_type=TransactionType.DEPOSIT.value, + description=body.description, + cost_center_id=body.cost_center_id, + ) + + +@router.post("/users/{user_id}/withdraw", response_model=WalletAdjustmentResponse) +async def withdraw_user_balance( + user_id: int, + body: WalletAdjustmentRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Withdraw funds from a user's wallet.""" + actor = await _require_authenticated_user(current_user) + await _get_user_or_404(db, user_id) + return await _create_wallet_adjustment( + db, + target_user_id=user_id, + actor_user_id=actor.id, + amount=-body.amount, + transaction_type=TransactionType.WITHDRAW.value, + description=body.description, + cost_center_id=body.cost_center_id, + ) + + +@router.post("/rebuild-balance-ledger") +async def rebuild_balance_ledger( + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Recompute balance_after for all wallet transactions. + + This rebuilds the running balance for all users and cost centers. + - Personal transactions: per-user running balance + - Cost-center transactions: global running balance for the entire cost center + """ + await _require_authenticated_user(current_user) + + # Get ALL transactions sorted by timestamp + result = await db.execute( + select(WalletTransaction).order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) + ) + all_transactions = result.scalars().all() + + # Build running balances per (user, cost_center_id) pair + # For each cost center, track its global running balance + # For each user's personal balance, track that separately + cc_running_balances: dict[int, float] = {} # cost_center_id -> running balance + user_personal_balances: dict[int, float] = {} # user_id -> personal running balance + + tx_updates: list[tuple[WalletTransaction, float]] = [] + + for tx in all_transactions: + if tx.cost_center_id is None: + # Personal transaction: per-user running balance + current = user_personal_balances.get(tx.user_id, 0.0) + new_balance = current + float(tx.amount) + user_personal_balances[tx.user_id] = new_balance + tx_updates.append((tx, new_balance)) + else: + # Cost-center transaction: global running balance for this cost center + current = cc_running_balances.get(tx.cost_center_id, 0.0) + new_balance = current + float(tx.amount) + cc_running_balances[tx.cost_center_id] = new_balance + tx_updates.append((tx, new_balance)) + + # Update all transactions with the new balance_after values + for tx, new_balance in tx_updates: + tx.balance_after = new_balance + db.add(tx) + + await db.flush() + await db.commit() + + return { + "status": "success", + "transactions_rebuilt": len(all_transactions), + "message": f"Rebuilt balance_after for {len(all_transactions)} transactions", + } + + +@router.get("/cost-centers", response_model=list[CostCenterSummaryResponse]) +async def list_cost_centers( + include_inactive: bool = Query(False), + db: AsyncSession = Depends(get_db), + current_user: User | None = Depends(require_auth_if_enabled), +): + """List all cost centers. + + Requires admin-level finance permissions. + """ + await _require_cost_center_admin_access(current_user) + query = select(CostCenter).order_by(CostCenter.is_private.desc(), CostCenter.name.asc()) + if not include_inactive: + query = query.where(CostCenter.is_active.is_(True)) + + result = await db.execute(query) + centers = result.scalars().all() + center_ids = [center.id for center in centers] + total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, center_ids) + total_balance_map = await _get_cost_center_balance_map(db, center_ids) + reserved_map = await _get_cost_center_reserved_map(db, center_ids) + + return [ + _to_cost_center_summary( + center, + can_print=True, + total_usage=total_usage_map.get(center.id, 0.0), + month_usage=month_usage_map.get(center.id, 0.0), + total_balance=total_balance_map.get(center.id, 0.0), + reserved=reserved_map.get(center.id, 0.0), + ) + for center in centers + ] + + +@router.post("/cost-centers", response_model=CostCenterSummaryResponse) +async def create_cost_center( + body: CostCenterCreateRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_CREATE), +): + """Create a shared cost center.""" + await _require_authenticated_user(current_user) + total_budget = body.total_budget + monthly_budget = body.monthly_budget + if monthly_budget is not None: + total_budget = None + elif total_budget is not None: + monthly_budget = None + + center = CostCenter( + name=body.name.strip(), + is_active=body.is_active, + is_private=False, + owner_user_id=None, + total_budget=total_budget, + monthly_budget=monthly_budget, + ) + db.add(center) + await db.flush() + await db.commit() + await db.refresh(center) + + return _to_cost_center_summary(center, can_print=True, total_usage=0.0, month_usage=0.0, total_balance=0.0) + + +@router.get("/cost-centers/{cost_center_id}", response_model=CostCenterDetailResponse) +async def get_cost_center( + cost_center_id: int, + db: AsyncSession = Depends(get_db), + current_user: User | None = Depends(require_auth_if_enabled), +): + """Get one cost center with its memberships.""" + await _require_cost_center_admin_access(current_user) + center = await _get_cost_center_or_404(db, cost_center_id) + total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, [center.id]) + total_balance_map = await _get_cost_center_balance_map(db, [center.id]) + reserved_map = await _get_cost_center_reserved_map(db, [center.id]) + summary = _to_cost_center_summary( + center, + can_print=True, + total_usage=total_usage_map.get(center.id, 0.0), + month_usage=month_usage_map.get(center.id, 0.0), + total_balance=total_balance_map.get(center.id, 0.0), + reserved=reserved_map.get(center.id, 0.0), + ) + return CostCenterDetailResponse( + **summary.model_dump(), + members=[CostCenterMemberResponse.model_validate(m) for m in center.members], + ) + + +@router.patch("/cost-centers/{cost_center_id}", response_model=CostCenterSummaryResponse) +async def update_cost_center( + cost_center_id: int, + body: CostCenterUpdateRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Update name or active-state of a cost center.""" + await _require_authenticated_user(current_user) + center = await _get_cost_center_or_404(db, cost_center_id) + + if body.name is not None: + center.name = body.name.strip() + if body.is_active is not None: + center.is_active = body.is_active + + await db.flush() + + total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, [center.id]) + total_balance_map = await _get_cost_center_balance_map(db, [center.id]) + reserved_map = await _get_cost_center_reserved_map(db, [center.id]) + return _to_cost_center_summary( + center, + can_print=True, + total_usage=total_usage_map.get(center.id, 0.0), + month_usage=month_usage_map.get(center.id, 0.0), + total_balance=total_balance_map.get(center.id, 0.0), + reserved=reserved_map.get(center.id, 0.0), + ) + + +@router.patch("/cost-centers/{cost_center_id}/budgets", response_model=CostCenterSummaryResponse) +async def update_cost_center_budgets( + cost_center_id: int, + body: CostCenterBudgetUpdateRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Update budget values of a cost center.""" + await _require_authenticated_user(current_user) + center = await _get_cost_center_or_404(db, cost_center_id) + + if body.monthly_budget is not None: + center.monthly_budget = body.monthly_budget + center.total_budget = None + elif body.total_budget is not None: + center.total_budget = body.total_budget + center.monthly_budget = None + else: + center.total_budget = None + center.monthly_budget = None + await db.flush() + + total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, [center.id]) + total_balance_map = await _get_cost_center_balance_map(db, [center.id]) + reserved_map = await _get_cost_center_reserved_map(db, [center.id]) + return _to_cost_center_summary( + center, + can_print=True, + total_usage=total_usage_map.get(center.id, 0.0), + month_usage=month_usage_map.get(center.id, 0.0), + total_balance=total_balance_map.get(center.id, 0.0), + reserved=reserved_map.get(center.id, 0.0), + ) + + +@router.post("/cost-centers/{cost_center_id}/members", response_model=CostCenterMemberResponse) +async def upsert_cost_center_member( + cost_center_id: int, + body: CostCenterMemberRequest, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Assign or update a user's membership on a cost center.""" + await _require_authenticated_user(current_user) + center = await _get_cost_center_or_404(db, cost_center_id) + + if center.is_private: + raise HTTPException(status_code=400, detail="Private cost center memberships cannot be modified") + + await _get_user_or_404(db, body.user_id) + + existing = await db.execute( + select(CostCenterMember).where( + CostCenterMember.cost_center_id == cost_center_id, + CostCenterMember.user_id == body.user_id, + ) + ) + member = existing.scalar_one_or_none() + if member is None: + member = CostCenterMember(cost_center_id=cost_center_id, user_id=body.user_id, can_print=body.can_print) + db.add(member) + else: + member.can_print = body.can_print + + await db.flush() + await db.commit() + return CostCenterMemberResponse.model_validate(member) + + +@router.delete("/cost-centers/{cost_center_id}") +async def delete_cost_center( + cost_center_id: int, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Delete a shared cost center.""" + await _require_authenticated_user(current_user) + center = await _get_cost_center_or_404(db, cost_center_id) + + if center.is_private: + raise HTTPException(status_code=400, detail="Private cost centers cannot be deleted") + + balance_map = await _get_cost_center_balance_map(db, [center.id]) + total_balance = balance_map.get(center.id, 0.0) + if abs(total_balance) > 1e-9: + raise HTTPException(status_code=400, detail="Cost center can only be deleted when balance is 0") + + await db.delete(center) + await db.commit() + return {"status": "success"} + + +@router.delete("/cost-centers/{cost_center_id}/members/{user_id}") +async def remove_cost_center_member( + cost_center_id: int, + user_id: int, + db: AsyncSession = Depends(get_db), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY), +): + """Remove a user from a shared cost center.""" + await _require_authenticated_user(current_user) + center = await _get_cost_center_or_404(db, cost_center_id) + if center.is_private: + raise HTTPException(status_code=400, detail="Private cost center memberships cannot be modified") + + result = await db.execute( + select(CostCenterMember).where( + CostCenterMember.cost_center_id == cost_center_id, + CostCenterMember.user_id == user_id, + ) + ) + member = result.scalar_one_or_none() + if member is None: + raise HTTPException(status_code=404, detail="Membership not found") + + await db.delete(member) + await db.commit() + return {"status": "success"} diff --git a/backend/app/api/routes/library.py b/backend/app/api/routes/library.py index ed76b70d5..73f00c118 100644 --- a/backend/app/api/routes/library.py +++ b/backend/app/api/routes/library.py @@ -65,6 +65,7 @@ from backend.app.schemas.library import ( ) from backend.app.schemas.slicer import SliceRequest, SliceResponse from backend.app.services.archive import ThreeMFParser +from backend.app.services.finance_budget import validate_print_budget from backend.app.services.plate_thumbnail import inject_plate_thumbnails_if_missing from backend.app.services.stl_thumbnail import MIN_USABLE_STL_BYTES, generate_stl_thumbnail from backend.app.utils.filename import InvalidFilenameError, validate_print_filename @@ -4181,6 +4182,13 @@ async def print_library_file( if not project_result.scalar_one_or_none(): raise HTTPException(status_code=404, detail="Project not found") + await validate_print_budget( + db, + cost_center_id=body.cost_center_id, + estimated_cost=body.estimated_cost, + current_user=current_user, + ) + plate_name = body.plate_name if not plate_name and body.plate_id is not None: plate_name = f"Plate {body.plate_id}" diff --git a/backend/app/api/routes/print_queue.py b/backend/app/api/routes/print_queue.py index f2c9534a4..7c7349196 100644 --- a/backend/app/api/routes/print_queue.py +++ b/backend/app/api/routes/print_queue.py @@ -36,6 +36,7 @@ from backend.app.schemas.print_queue import ( PrintQueueReorder, ) from backend.app.services.filament_deficit import compute_deficit_for_queue_item +from backend.app.services.finance_budget import validate_print_budget from backend.app.services.notification_service import notification_service from backend.app.utils.printer_models import normalize_printer_model, normalize_printer_model_id from backend.app.utils.threemf_tools import ( @@ -164,6 +165,8 @@ def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse: "waiting_reason": item.waiting_reason, "archive_id": item.archive_id, "library_file_id": item.library_file_id, + "cost_center_id": item.cost_center_id, + "estimated_cost": item.estimated_cost, "position": item.position, "scheduled_time": item.scheduled_time, "require_previous_success": item.require_previous_success, @@ -545,6 +548,14 @@ async def add_to_queue( if not project_result.scalar_one_or_none(): raise HTTPException(status_code=404, detail="Project not found") + await validate_print_budget( + db, + cost_center_id=data.cost_center_id, + estimated_cost=data.estimated_cost, + current_user=current_user, + quantity=quantity, + ) + ams_mapping_json = json.dumps(data.ams_mapping) if data.ams_mapping else None items = [] for i in range(quantity): @@ -556,6 +567,8 @@ async def add_to_queue( filament_overrides=filament_overrides_json, archive_id=data.archive_id, library_file_id=data.library_file_id, + cost_center_id=data.cost_center_id, + estimated_cost=data.estimated_cost, scheduled_time=data.scheduled_time, require_previous_success=data.require_previous_success, auto_off_after=data.auto_off_after, @@ -672,6 +685,7 @@ async def bulk_update_queue_items( updated_count = 0 skipped_count = 0 + validates_billing_fields = "cost_center_id" in update_data or "estimated_cost" in update_data for item in items: if item.status != "pending": @@ -683,6 +697,15 @@ async def bulk_update_queue_items( skipped_count += 1 continue + if validates_billing_fields: + await validate_print_budget( + db, + cost_center_id=update_data.get("cost_center_id", item.cost_center_id), + estimated_cost=update_data.get("estimated_cost", item.estimated_cost), + current_user=user, + exclude_queue_item_id=item.id, + ) + for field, value in update_data.items(): setattr(item, field, value) updated_count += 1 @@ -1022,12 +1045,13 @@ async def update_queue_item( json.dumps(update_data["filament_overrides"]) if update_data["filament_overrides"] else None ) - # Serialize H2C rack-swap nozzle pick (#1780) to JSON for TEXT column - # storage; same Text-as-opaque-blob convention as ams_mapping above. - if "nozzle_mapping" in update_data: - update_data["nozzle_mapping"] = ( - json.dumps(update_data["nozzle_mapping"]) if update_data["nozzle_mapping"] else None - ) + await validate_print_budget( + db, + cost_center_id=update_data.get("cost_center_id", item.cost_center_id), + estimated_cost=update_data.get("estimated_cost", item.estimated_cost), + current_user=user, + exclude_queue_item_id=item.id, + ) for field, value in update_data.items(): setattr(item, field, value) @@ -1212,7 +1236,7 @@ async def start_queue_item( item_id: int, skip_filament_check: bool = Query(default=False), db: AsyncSession = Depends(get_db), - user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN), + current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN), ): """Manually start a staged (manual_start) queue item. @@ -1240,6 +1264,14 @@ async def start_queue_item( if item.status != "pending": raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'") + await validate_print_budget( + db, + cost_center_id=item.cost_center_id, + estimated_cost=item.estimated_cost, + current_user=current_user, + exclude_queue_item_id=item.id, + ) + # Live deficit check — re-evaluated against current spool state, so a # spool swap between scheduler flagging and the user clicking ▶ clears # the block automatically. @@ -1269,8 +1301,8 @@ async def start_queue_item( # (#1670). An item that already has a creator (UI-added queue items) # keeps that attribution; the dispatcher is not promoted over the # original uploader. - if user is not None and item.created_by_id is None: - item.created_by_id = user.id + if current_user is not None and item.created_by_id is None: + item.created_by_id = current_user.id await db.commit() await db.refresh(item, ["archive", "printer", "library_file", "created_by", "batch"]) diff --git a/backend/app/api/routes/settings.py b/backend/app/api/routes/settings.py index 3143dc123..5be40abce 100644 --- a/backend/app/api/routes/settings.py +++ b/backend/app/api/routes/settings.py @@ -135,6 +135,8 @@ async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) - "default_vibration_cali", "default_layer_inspect", "default_timelapse", + "billing_enabled", + "printer_kill_switch_enabled", "default_nozzle_offset_cali", "ldap_enabled", "ldap_auto_provision", @@ -161,6 +163,7 @@ async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) - "stagger_group_size", "stagger_interval_minutes", "forecast_global_lead_time_days", + "finance_budget_reset_day", "session_max_hours", ]: settings_dict[setting.key] = int(setting.value) diff --git a/backend/app/api/routes/users.py b/backend/app/api/routes/users.py index 0cbed1ba7..1eec35380 100644 --- a/backend/app/api/routes/users.py +++ b/backend/app/api/routes/users.py @@ -41,6 +41,7 @@ from backend.app.services.email_service import ( get_smtp_settings, send_email, ) +from backend.app.services.finance_defaults import ensure_user_finance_defaults router = APIRouter(prefix="/users", tags=["users"]) @@ -164,6 +165,8 @@ async def create_user( new_user.groups = list(groups) db.add(new_user) + await db.flush() + await ensure_user_finance_defaults(db, new_user) await db.commit() await db.refresh(new_user) @@ -307,6 +310,8 @@ async def update_user( ) user.groups = list(groups) + await ensure_user_finance_defaults(db, user) + await db.commit() result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups))) user = result.scalar_one() diff --git a/backend/app/main.py b/backend/app/main.py index 942efc581..837cd55eb 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -30,6 +30,7 @@ from backend.app.api.routes import ( discovery, external_links, filaments, + finance, firmware, github_backup, groups, @@ -372,6 +373,9 @@ _expected_prints: dict[tuple[int, str], int] = {} # Used by usage tracker to map 3MF slots to physical AMS trays _print_ams_mappings: dict[int, list[int]] = {} +# Track cost center selection for the current print run: {archive_id: cost_center_id} +_print_cost_center_ids: dict[int, int] = {} + # Track plate_id for prints from multi-plate 3MFs: {archive_id: plate_id} # Used by usage tracker to scope 3MF parsing to the dispatched plate (#1697). # Populated by direct-Print and queue dispatch paths; queue prints also have a @@ -386,6 +390,9 @@ _last_progress_milestone: dict[int, int] = {} # Track whether first layer complete notification has been sent for current print _first_layer_notified: dict[int, bool] = {} +# Track whether we already sent a kill-switch stop for the current unauthorized print +_unauthorized_print_kill_sent: set[int] = set() + # Track HMS errors that have been notified: {printer_id: set of error codes} # This prevents sending duplicate notifications for the same error _notified_hms_errors: dict[int, set[str]] = {} @@ -611,6 +618,54 @@ _expected_print_registered_at: dict[tuple[int, str], float] = {} _EXPECTED_PRINT_CLEANUP_INTERVAL: int = 15 * 60 # 15 minutes _expected_prints_cleanup_task: asyncio.Task | None = None +_ACTIVE_PRINT_STATES: set[str] = {"RUNNING", "PRINTING", "PAUSE"} + + +def _build_status_print_keys(printer_id: int, state: PrinterState) -> list[tuple[int, str]]: + """Build filename keys for matching a printer status update to Bambuddy-owned jobs.""" + + possible_keys: list[tuple[int, str]] = [] + filename = (state.gcode_file or state.current_print or "").strip() + subtask_name = (state.subtask_name or "").strip() + + if subtask_name: + possible_keys.append((printer_id, subtask_name)) + possible_keys.append((printer_id, f"{subtask_name}.3mf")) + possible_keys.append((printer_id, f"{subtask_name}.gcode.3mf")) + + if filename: + base_name = filename.rsplit("/", 1)[-1] + if base_name.endswith(".gcode.3mf"): + root_name = base_name[: -len(".gcode.3mf")] + possible_keys.append((printer_id, root_name)) + possible_keys.append((printer_id, base_name)) + possible_keys.append((printer_id, f"{root_name}.gcode")) + possible_keys.append((printer_id, f"{root_name}.3mf")) + elif base_name.endswith(".3mf"): + root_name = base_name[: -len(".3mf")] + possible_keys.append((printer_id, root_name)) + possible_keys.append((printer_id, base_name)) + elif base_name.endswith(".gcode"): + root_name = base_name[: -len(".gcode")] + possible_keys.append((printer_id, root_name)) + possible_keys.append((printer_id, f"{root_name}.3mf")) + possible_keys.append((printer_id, base_name)) + else: + possible_keys.append((printer_id, base_name)) + possible_keys.append((printer_id, f"{base_name}.3mf")) + + return possible_keys + + +def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState) -> bool: + """Return True when the current status belongs to a print started by Bambuddy.""" + + if printer_manager.get_current_print_user(printer_id): + return True + + possible_keys = _build_status_print_keys(printer_id, state) + return any(key in _expected_prints or key in _active_prints for key in possible_keys) + async def _get_plug_energy(plug, db) -> dict | None: """Get energy from plug regardless of type (Tasmota, Home Assistant, MQTT, or REST). @@ -683,6 +738,7 @@ def register_expected_print( archive_id: int, ams_mapping: list[int] | None = None, created_by_id: int | None = None, + cost_center_id: int | None = None, plate_id: int | None = None, ): """Register an expected print from reprint/scheduled so we don't create duplicate archives.""" @@ -696,6 +752,8 @@ def register_expected_print( # Store AMS mapping for usage tracking at print completion if ams_mapping is not None: _print_ams_mappings[archive_id] = ams_mapping + if cost_center_id is not None: + _print_cost_center_ids[archive_id] = cost_center_id # Store plate_id for usage tracking when this is a single-plate dispatch from # a multi-plate 3MF — without this, the direct-Print path attributes the whole # file's filament total to the spool instead of just the printed plate (#1697). @@ -1184,6 +1242,45 @@ async def on_printer_status_change(printer_id: int, state: PrinterState): f"{ams_dry_key}:{ams_tray_key}:{state.door_open}:{state.ams_filament_backup}" ) + is_active_print = state.state in _ACTIVE_PRINT_STATES + if not is_active_print: + _unauthorized_print_kill_sent.discard(printer_id) + else: + kill_switch_enabled = False + status_logger = logging.getLogger(__name__) + try: + async with async_session() as db: + from backend.app.services.finance_budget import is_printer_kill_switch_enabled + + kill_switch_enabled = await is_printer_kill_switch_enabled(db) + except Exception as e: + status_logger.warning("[KILL SWITCH] Failed to read kill-switch setting for printer %s: %s", printer_id, e) + + if not kill_switch_enabled or _is_bambuddy_authorized_print(printer_id, state): + _unauthorized_print_kill_sent.discard(printer_id) + elif printer_id in _unauthorized_print_kill_sent: + pass + else: + try: + stopped = printer_manager.stop_print(printer_id) + if stopped: + _unauthorized_print_kill_sent.add(printer_id) + status_logger.warning( + "[KILL SWITCH] Stopped unauthorized print on printer %s (state=%s)", + printer_id, + state.state, + ) + else: + status_logger.warning( + "[KILL SWITCH] Could not stop unauthorized print on printer %s (state=%s)", + printer_id, + state.state, + ) + except Exception as e: + status_logger.warning( + "[KILL SWITCH] Failed to stop unauthorized print on printer %s: %s", printer_id, e + ) + # MQTT relay - publish status (before dedup check - always publish to MQTT) try: printer_info = printer_manager.get_printer(printer_id) @@ -4603,6 +4700,29 @@ async def on_print_complete(printer_id: int, data: dict): log_timing("Archive status update") + # Apply finance wallet charge or release reservations once + try: + if data.get("status") in ("completed", "failed", "aborted", "cancelled"): + async with async_session() as db: + from backend.app.models.archive import PrintArchive + from backend.app.services.finance_billing import apply_print_charge_for_archive + + archive = await db.get(PrintArchive, archive_id) + cost_center_id = _print_cost_center_ids.pop(archive_id, None) + charged = await apply_print_charge_for_archive( + db, + archive_id, + cost_center_id=cost_center_id, + print_run_id=archive.subtask_id if archive else None, + ) + await db.commit() + if charged: + logger.info("[FINANCE] Applied print charge for archive %s", archive_id) + except Exception as e: + logger.warning("[FINANCE] Failed to apply print charge for archive %s: %s", archive_id, e) + + log_timing("Finance charge update") + # Write independent print log entry (separate table, never touches archives) try: async with async_session() as db: @@ -5830,6 +5950,7 @@ def _evict_stale_expected_prints() -> None: for archive_id in evicted_archive_ids: if archive_id not in live_archive_ids: _print_ams_mappings.pop(archive_id, None) + _print_cost_center_ids.pop(archive_id, None) _print_plate_ids.pop(archive_id, None) logging.getLogger(__name__).info( @@ -6711,6 +6832,7 @@ app.include_router(groups.router, prefix=app_settings.api_prefix) app.include_router(printers.router, prefix=app_settings.api_prefix) app.include_router(archives.router, prefix=app_settings.api_prefix) app.include_router(filaments.router, prefix=app_settings.api_prefix) +app.include_router(finance.router, prefix=app_settings.api_prefix) app.include_router(inventory.router, prefix=app_settings.api_prefix) app.include_router(labels.router, prefix=app_settings.api_prefix) app.include_router(settings_routes.router, prefix=app_settings.api_prefix) diff --git a/backend/app/models/archive.py b/backend/app/models/archive.py index e73903293..c705df01f 100644 --- a/backend/app/models/archive.py +++ b/backend/app/models/archive.py @@ -12,6 +12,9 @@ class PrintArchive(Base): id: Mapped[int] = mapped_column(primary_key=True) printer_id: Mapped[int | None] = mapped_column(ForeignKey("printers.id"), nullable=True) project_id: Mapped[int | None] = mapped_column(ForeignKey("projects.id", ondelete="SET NULL"), nullable=True) + cost_center_id: Mapped[int | None] = mapped_column( + ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True + ) # File info filename: Mapped[str] = mapped_column(String(255)) @@ -68,6 +71,7 @@ class PrintArchive(Base): # User additions is_favorite: Mapped[bool] = mapped_column(Boolean, default=False) + wallet_charge_skipped: Mapped[bool] = mapped_column(Boolean, default=False) tags: Mapped[str | None] = mapped_column(Text) notes: Mapped[str | None] = mapped_column(Text) cost: Mapped[float | None] = mapped_column(Float) @@ -98,9 +102,11 @@ class PrintArchive(Base): # Relationships printer: Mapped["Printer | None"] = relationship(back_populates="archives") project: Mapped["Project | None"] = relationship(back_populates="archives") + cost_center: Mapped["CostCenter | None"] = relationship() created_by: Mapped["User | None"] = relationship() +from backend.app.models.finance import CostCenter # noqa: E402, F811 from backend.app.models.printer import Printer # noqa: E402, F811 from backend.app.models.project import Project # noqa: E402, F811 from backend.app.models.user import User # noqa: E402, F811 diff --git a/backend/app/models/finance.py b/backend/app/models/finance.py new file mode 100644 index 000000000..632aa8335 --- /dev/null +++ b/backend/app/models/finance.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import uuid +from datetime import datetime +from enum import Enum as PyEnum +from typing import TYPE_CHECKING + +from sqlalchemy import Boolean, CheckConstraint, DateTime, Float, ForeignKey, String, Text, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column, relationship, validates + +from backend.app.core.database import Base + +if TYPE_CHECKING: + from backend.app.models.archive import PrintArchive + from backend.app.models.print_queue import PrintQueueItem + from backend.app.models.user import User + + +class TransactionType(str, PyEnum): + PRINT_CHARGE = "print_charge" + DEPOSIT = "deposit" + WITHDRAW = "withdraw" + MANUAL_ADJUSTMENT = "manual_adjustment" + + +VALID_TRANSACTION_TYPES = {item.value for item in TransactionType} + + +def normalize_transaction_type(value: str | TransactionType) -> str: + if isinstance(value, TransactionType): + return value.value + if value not in VALID_TRANSACTION_TYPES: + raise ValueError(f"Invalid transaction type: {value}") + return value + + +class UserWallet(Base): + """Per-user wallet balance. + + Balance updates are driven by wallet transactions. + """ + + __tablename__ = "user_wallets" + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), unique=True, index=True) + balance: Mapped[float] = mapped_column(Float, default=0.0) + currency: Mapped[str] = mapped_column(String(3), default="EUR") + updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) + + user: Mapped[User] = relationship() + + +class CostCenter(Base): + """Cost center for assigning print costs and budgets.""" + + __tablename__ = "cost_centers" + + id: Mapped[int] = mapped_column(primary_key=True) + code: Mapped[str] = mapped_column(String(32), unique=True, index=True, default=lambda: uuid.uuid4().hex[:12]) + name: Mapped[str] = mapped_column(String(150), index=True) + is_active: Mapped[bool] = mapped_column(Boolean, default=True) + is_private: Mapped[bool] = mapped_column(Boolean, default=False) + owner_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True) + + total_budget: Mapped[float | None] = mapped_column(Float, nullable=True) + monthly_budget: Mapped[float | None] = mapped_column(Float, nullable=True) + + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now()) + + owner: Mapped[User | None] = relationship() + members: Mapped[list[CostCenterMember]] = relationship( + "CostCenterMember", + back_populates="cost_center", + cascade="all, delete-orphan", + lazy="selectin", + ) + + +class CostCenterMember(Base): + """User-to-cost-center assignment with print permission.""" + + __tablename__ = "cost_center_members" + __table_args__ = (UniqueConstraint("cost_center_id", "user_id", name="uq_cost_center_members_cc_user"),) + + id: Mapped[int] = mapped_column(primary_key=True) + cost_center_id: Mapped[int] = mapped_column(ForeignKey("cost_centers.id", ondelete="CASCADE"), index=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) + can_print: Mapped[bool] = mapped_column(Boolean, default=True) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + + cost_center: Mapped[CostCenter] = relationship("CostCenter", back_populates="members") + user: Mapped[User] = relationship() + + +class BudgetReservation(Base): + """Persisted budget hold for accepted print work that has not been charged yet.""" + + __tablename__ = "budget_reservations" + + id: Mapped[int] = mapped_column(primary_key=True) + cost_center_id: Mapped[int] = mapped_column(ForeignKey("cost_centers.id", ondelete="CASCADE"), index=True) + amount: Mapped[float] = mapped_column(Float) + status: Mapped[str] = mapped_column(String(20), default="active", index=True) + source_type: Mapped[str] = mapped_column(String(50), index=True) + source_id: Mapped[int | None] = mapped_column(index=True) + print_archive_id: Mapped[int | None] = mapped_column( + ForeignKey("print_archives.id", ondelete="SET NULL"), nullable=True, index=True + ) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + released_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + + cost_center: Mapped[CostCenter] = relationship() + print_archive: Mapped[PrintArchive | None] = relationship() + + +class WalletTransaction(Base): + """Immutable wallet ledger entry.""" + + __tablename__ = "wallet_transactions" + __table_args__ = ( + CheckConstraint( + "transaction_type IN ('print_charge', 'deposit', 'withdraw', 'manual_adjustment')", + name="ck_wallet_transactions_transaction_type", + ), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) + cost_center_id: Mapped[int | None] = mapped_column( + ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True, index=True + ) + + transaction_type: Mapped[str] = mapped_column(String(40), index=True) + amount: Mapped[float] = mapped_column(Float) + balance_after: Mapped[float | None] = mapped_column(Float, nullable=True) + description: Mapped[str | None] = mapped_column(Text, nullable=True) + + created_by_user_id: Mapped[int | None] = mapped_column( + ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True + ) + print_run_id: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True) + print_archive_id: Mapped[int | None] = mapped_column( + ForeignKey("print_archives.id", ondelete="SET NULL"), nullable=True, index=True + ) + print_queue_id: Mapped[int | None] = mapped_column( + ForeignKey("print_queue.id", ondelete="SET NULL"), nullable=True, index=True + ) + + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), index=True) + + user: Mapped[User] = relationship(foreign_keys=[user_id]) + cost_center: Mapped[CostCenter | None] = relationship() + created_by: Mapped[User | None] = relationship(foreign_keys=[created_by_user_id]) + print_archive: Mapped[PrintArchive | None] = relationship() + print_queue: Mapped[PrintQueueItem | None] = relationship() + + @validates("transaction_type") + def _validate_transaction_type(self, key: str, value: str | TransactionType) -> str: + return normalize_transaction_type(value) diff --git a/backend/app/models/print_queue.py b/backend/app/models/print_queue.py index bc4a7b771..b0928de51 100644 --- a/backend/app/models/print_queue.py +++ b/backend/app/models/print_queue.py @@ -1,6 +1,6 @@ from datetime import datetime -from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, func +from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, Text, func from sqlalchemy.orm import Mapped, mapped_column, relationship from backend.app.core.database import Base @@ -32,6 +32,10 @@ class PrintQueueItem(Base): library_file_id: Mapped[int | None] = mapped_column( ForeignKey("library_files.id", ondelete="CASCADE"), nullable=True ) + cost_center_id: Mapped[int | None] = mapped_column( + ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True + ) + estimated_cost: Mapped[float | None] = mapped_column(Float, nullable=True) project_id: Mapped[int | None] = mapped_column(ForeignKey("projects.id", ondelete="SET NULL"), nullable=True) batch_id: Mapped[int | None] = mapped_column(ForeignKey("print_batches.id", ondelete="SET NULL"), nullable=True) @@ -120,12 +124,14 @@ class PrintQueueItem(Base): printer: Mapped["Printer"] = relationship() archive: Mapped["PrintArchive | None"] = relationship() library_file: Mapped["LibraryFile | None"] = relationship() + cost_center: Mapped["CostCenter | None"] = relationship() project: Mapped["Project | None"] = relationship(back_populates="queue_items") batch: Mapped["PrintBatch | None"] = relationship(back_populates="queue_items") created_by: Mapped["User | None"] = relationship() from backend.app.models.archive import PrintArchive # noqa: E402 +from backend.app.models.finance import CostCenter # noqa: E402 from backend.app.models.library import LibraryFile # noqa: E402 from backend.app.models.print_batch import PrintBatch # noqa: E402 from backend.app.models.printer import Printer # noqa: E402 diff --git a/backend/app/services/archive.py b/backend/app/services/archive.py index 2702bc84a..9c007272b 100644 --- a/backend/app/services/archive.py +++ b/backend/app/services/archive.py @@ -1127,6 +1127,7 @@ class ArchiveService: created_by_id: int | None = None, original_filename: str | None = None, project_id: int | None = None, + cost_center_id: int | None = None, subtask_id: str | None = None, prefer_filename_for_name: bool = False, ) -> PrintArchive | None: @@ -1300,6 +1301,7 @@ class ArchiveService: extra_data=metadata, created_by_id=created_by_id, project_id=project_id, + cost_center_id=cost_center_id, subtask_id=subtask_id, ) diff --git a/backend/app/services/background_dispatch.py b/backend/app/services/background_dispatch.py index 582a5106b..0fef9ed96 100644 --- a/backend/app/services/background_dispatch.py +++ b/backend/app/services/background_dispatch.py @@ -22,8 +22,10 @@ from backend.app.core.config import settings from backend.app.core.database import async_session from backend.app.core.tasks import spawn_background_task from backend.app.core.websocket import ws_manager +from backend.app.models.finance import BudgetReservation, CostCenter from backend.app.models.library import LibraryFile from backend.app.models.printer import Printer +from backend.app.models.user import User from backend.app.services.archive import ArchiveService from backend.app.services.bambu_ftp import ( cache_3mf_download, @@ -32,6 +34,7 @@ from backend.app.services.bambu_ftp import ( upload_file_async, with_ftp_retry, ) +from backend.app.services.finance_budget import create_budget_reservation, release_budget_reservation from backend.app.services.printer_manager import printer_manager from backend.app.utils.filename import derive_remote_filename @@ -116,12 +119,16 @@ class BackgroundDispatchService: dispatcher = self._dispatcher_task self._dispatcher_task = None running_tasks = list(self._running_tasks.values()) + jobs_to_release = [*self._queued_jobs, *(state.job for state in self._active_jobs.values())] self._running_tasks.clear() self._active_jobs.clear() self._queued_jobs.clear() self._cancel_requested_job_ids.clear() self._job_event.set() + for job in jobs_to_release: + await self._release_budget_reservation(job, status="released") + if dispatcher: dispatcher.cancel() for task in running_tasks: @@ -290,8 +297,22 @@ class BackgroundDispatchService: raise DispatchEnqueueRejected(f"Printer {printer_name} is currently busy printing") dispatch_position = len(self._queued_jobs) + len(self._active_jobs) + 1 + job_id = self._next_job_id + async with async_session() as db: + requested_by = await db.get(User, requested_by_user_id) if requested_by_user_id is not None else None + await create_budget_reservation( + db, + cost_center_id=options.get("cost_center_id"), + estimated_cost=options.get("estimated_cost"), + current_user=requested_by, + source_type="background_dispatch", + source_id=job_id, + print_archive_id=source_id if kind == "reprint_archive" else None, + ) + await db.commit() + job = PrintDispatchJob( - id=self._next_job_id, + id=job_id, kind=kind, source_id=source_id, source_name=source_name, @@ -431,6 +452,9 @@ class BackgroundDispatchService: await ws_manager.broadcast({"type": "background_dispatch", "data": payload}) async def _mark_job_finished(self, job: PrintDispatchJob, *, failed: bool, message: str): + if failed: + await self._release_budget_reservation(job, status="released") + async with self._lock: if failed: self._batch_failed += 1 @@ -463,6 +487,8 @@ class BackgroundDispatchService: self._batch_failed = 0 async def _mark_job_cancelled(self, job: PrintDispatchJob): + await self._release_budget_reservation(job, status="released") + async with self._lock: self._active_jobs.pop(job.id, None) self._running_tasks.pop(job.id, None) @@ -493,6 +519,19 @@ class BackgroundDispatchService: if self._is_cancel_requested(job.id): raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled") + @staticmethod + async def _release_budget_reservation(job: PrintDispatchJob, *, status: str): + if job.options.get("cost_center_id") is None: + return + async with async_session() as db: + await release_budget_reservation( + db, + source_type="background_dispatch", + source_id=job.id, + status=status, + ) + await db.commit() + def _build_state_payload_unlocked(self, recent_event: dict[str, Any] | None = None) -> dict[str, Any]: processing = len(self._active_jobs) dispatched = len(self._queued_jobs) @@ -565,6 +604,12 @@ class BackgroundDispatchService: if not archive: raise RuntimeError("Archive not found") + cost_center_id = job.options.get("cost_center_id") + if cost_center_id is not None: + cost_center = await db.scalar(select(CostCenter).where(CostCenter.id == cost_center_id)) + if not cost_center: + raise RuntimeError("Cost center not found") + printer = await db.scalar(select(Printer).where(Printer.id == job.printer_id)) if not printer: raise RuntimeError("Printer not found") @@ -669,6 +714,7 @@ class BackgroundDispatchService: remote_filename, job.source_id, ams_mapping=job.options.get("ams_mapping"), + cost_center_id=job.options.get("cost_center_id"), plate_id=plate_id, ) @@ -783,11 +829,21 @@ class BackgroundDispatchService: original_filename=lib_file.filename, project_id=job.project_id, created_by_id=job.requested_by_user_id, + cost_center_id=job.options.get("cost_center_id"), ) if not archive: raise RuntimeError("Failed to create archive") - - await db.flush() + if job.options.get("cost_center_id") is not None: + reservation = await db.scalar( + select(BudgetReservation).where( + BudgetReservation.source_type == "background_dispatch", + BudgetReservation.source_id == job.id, + BudgetReservation.status == "active", + ) + ) + if reservation: + reservation.print_archive_id = archive.id + await db.flush() remote_filename = derive_remote_filename(lib_file.filename) remote_path = f"/{remote_filename}" @@ -876,6 +932,7 @@ class BackgroundDispatchService: remote_filename, archive.id, ams_mapping=job.options.get("ams_mapping"), + cost_center_id=job.options.get("cost_center_id"), plate_id=plate_id, ) diff --git a/backend/app/services/finance_billing.py b/backend/app/services/finance_billing.py new file mode 100644 index 000000000..6dafca64d --- /dev/null +++ b/backend/app/services/finance_billing.py @@ -0,0 +1,213 @@ +import logging + +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError, SQLAlchemyError +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.app.models.archive import PrintArchive +from backend.app.models.finance import TransactionType, UserWallet, WalletTransaction +from backend.app.services.finance_budget import is_billing_enabled, release_budget_reservation + +logger = logging.getLogger(__name__) + + +async def _get_balance_after_for_transaction( + db: AsyncSession, + user_id: int, + cost_center_id: int | None, + amount: float, +) -> float: + """Calculate balance_after for a transaction. + + For cost-center transactions: sum of ALL transactions for that cost center (global). + For personal transactions (cost_center_id=None): user's wallet balance (personal). + + Args: + user_id: The user making the transaction + cost_center_id: The cost center (None for personal) + amount: The transaction amount (positive/negative) + + Returns: + The balance after this transaction would be applied + """ + try: + if cost_center_id is None: + # Personal transaction: use user wallet balance + wallet = (await db.execute(select(UserWallet).where(UserWallet.user_id == user_id))).scalar_one_or_none() + if wallet is None: + return float(amount) + return float(wallet.balance) + amount + else: + # Cost-center transaction: sum of ALL transactions for this cost center (global, not per-user) + result = await db.execute( + select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)).where( + WalletTransaction.cost_center_id == cost_center_id, + ) + ) + current_balance = float(result.scalar() or 0.0) + return current_balance + amount + except SQLAlchemyError as e: + logger.error(f"Database error in _get_balance_after_for_transaction: {e}", exc_info=True) + raise + + +def _calculate_partial_charge( + archive: PrintArchive, + base_cost: float, +) -> tuple[float, str]: + """Calculate proportional charge for partial prints based on filament usage. + + Returns (charge_amount, description_suffix) where: + - charge_amount: absolute cost to charge (0 if insufficient data) + - description_suffix: reason/details for transaction description + """ + try: + # Only apply proportional calculation for non-completed prints + if archive.status == "completed": + return round(float(base_cost), 2), "" + + filament_used = float(archive.filament_used_grams or 0.0) + filament_planned = None + + if archive.extra_data and isinstance(archive.extra_data, dict): + filament_planned = archive.extra_data.get("filament_grams_total") + if filament_planned is not None: + filament_planned = float(filament_planned) + + # If we don't have reliable planned filament data, do not guess a partial charge. + # Charging a failed/aborted print without an estimated baseline can overcharge users. + if filament_planned is None or filament_planned <= 0: + return 0.0, f"[{archive.status}: insufficient filament data]" + + # Calculate proportional cost + filament_ratio = min(1.0, max(0.0, filament_used / filament_planned)) # Clamp to [0, 1] + charge = float(base_cost) * filament_ratio + + # Round charges to 2 decimals for consistent persistence + charge = round(charge, 2) + + suffix = f"[{archive.status}: {filament_ratio:.1%} filament ({filament_used:.1f}g/{filament_planned:.1f}g)]" + return charge, suffix + except ValueError as e: + logger.error(f"Value error in _calculate_partial_charge: {e}", exc_info=True) + raise + + +async def apply_print_charge_for_archive( + db: AsyncSession, + archive_id: int, + *, + cost_center_id: int | None = None, + print_run_id: str | None = None, +) -> bool: + """Apply an idempotent wallet charge for a print archive. + + Charges completed prints at full cost, and partial/failed prints proportionally + based on actual filament used vs. planned filament. + + Returns True when a new wallet transaction was created. + """ + try: + if not await is_billing_enabled(db): + await release_budget_reservation(db, print_archive_id=archive_id, status="released") + logger.info("Billing is disabled; skipping print charge for archive ID %s.", archive_id) + return False + + archive = ( + await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id).with_for_update()) + ).scalar_one_or_none() + if archive is None: + logger.warning(f"Archive with ID {archive_id} not found.") + return False + + if archive.wallet_charge_skipped: + logger.info(f"Wallet charge skipped for archive ID {archive_id}.") + return False + + # Accept completed, aborted, cancelled, and failed prints + if archive.status not in ("completed", "aborted", "cancelled", "failed"): + logger.info(f"Archive ID {archive_id} has status {archive.status}, which is not chargeable.") + return False + + if archive.created_by_id is None: + logger.warning(f"Archive ID {archive_id} has no creator ID.") + return False + + base_cost = float(archive.cost or 0.0) + if base_cost <= 0: + logger.info(f"Base cost for archive ID {archive_id} is zero or negative.") + return False + + tx_conditions = [WalletTransaction.transaction_type == TransactionType.PRINT_CHARGE.value] + if print_run_id: + tx_conditions.append(WalletTransaction.print_run_id == print_run_id) + else: + tx_conditions.append(WalletTransaction.print_archive_id == archive.id) + + existing_tx = (await db.execute(select(WalletTransaction).where(*tx_conditions))).scalar_one_or_none() + if existing_tx is not None: + logger.info(f"Transaction already exists for archive ID {archive_id}.") + return False + + # Calculate charge (full for completed, partial for others) + charge, reason_suffix = _calculate_partial_charge(archive, base_cost) + if charge <= 0: + await release_budget_reservation(db, print_archive_id=archive.id, status="released") + logger.info(f"Calculated charge for archive ID {archive_id} is zero or negative.") + return False + + actual_cost_center_id = cost_center_id if cost_center_id is not None else archive.cost_center_id + + wallet = ( + await db.execute(select(UserWallet).where(UserWallet.user_id == archive.created_by_id)) + ).scalar_one_or_none() + if wallet is None: + wallet = UserWallet(user_id=archive.created_by_id, balance=0.0, currency="EUR") + db.add(wallet) + await db.flush() + logger.info(f"Created new wallet for user ID {archive.created_by_id}.") + + # Persist wallet balances rounded to cents + new_wallet_balance = round(float(wallet.balance) - charge, 2) + wallet.balance = new_wallet_balance + + label = archive.print_name or archive.filename or f"Archive {archive.id}" + description = f"Print charge: {label}{' ' + reason_suffix if reason_suffix else ''}" + + balance_after = await _get_balance_after_for_transaction( + db, archive.created_by_id, actual_cost_center_id, -charge + ) + if balance_after is not None: + balance_after = round(float(balance_after), 2) + + tx = WalletTransaction( + user_id=archive.created_by_id, + cost_center_id=actual_cost_center_id, + transaction_type=TransactionType.PRINT_CHARGE.value, + amount=-charge, + balance_after=balance_after, + description=description, + created_by_user_id=None, + print_run_id=print_run_id or archive.subtask_id, + print_archive_id=archive.id, + ) + db.add(tx) + # Ensure the transaction is flushed to detect unique/index constraint violations + try: + await db.flush() + except IntegrityError as e: + # Another concurrent worker likely created the same transaction + logger.info("Transaction already exists for archive ID %s (concurrent), skipping: %s", archive_id, e) + await db.rollback() + return False + + # Consume matching budget reservations after the transaction is persisted + await release_budget_reservation(db, print_archive_id=archive.id, status="consumed") + logger.info(f"Applied print charge for archive ID {archive_id}. New balance: {new_wallet_balance}.") + return True + except SQLAlchemyError as e: + logger.error(f"Database error in apply_print_charge_for_archive: {e}", exc_info=True) + return False + except ValueError as e: + logger.error(f"Value error in apply_print_charge_for_archive: {e}", exc_info=True) + return False diff --git a/backend/app/services/finance_budget.py b/backend/app/services/finance_budget.py new file mode 100644 index 000000000..444be3f06 --- /dev/null +++ b/backend/app/services/finance_budget.py @@ -0,0 +1,234 @@ +"""Budget validation helpers for finance-aware print dispatch.""" + +import calendar +from datetime import datetime, timezone +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from fastapi import HTTPException +from sqlalchemy import case, func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.app.models.finance import BudgetReservation, CostCenter, CostCenterMember, WalletTransaction +from backend.app.models.print_queue import PrintQueueItem +from backend.app.models.settings import Settings +from backend.app.models.user import User + + +async def is_billing_enabled(db: AsyncSession) -> bool: + # Consider any 'billing_enabled' setting with a true-ish value as enabling billing. + result = await db.execute( + select(func.count()) + .select_from(Settings) + .where(Settings.key == "billing_enabled", func.lower(func.coalesce(Settings.value, "")) == "true") + ) + count = int(result.scalar_one() or 0) + return count > 0 + + +async def is_printer_kill_switch_enabled(db: AsyncSession) -> bool: + """Return True when billing and the printer kill-switch are both enabled.""" + + result = await db.execute( + select(Settings.key, Settings.value).where(Settings.key.in_(("billing_enabled", "printer_kill_switch_enabled"))) + ) + values = {key: (value or "").strip().lower() for key, value in result.all()} + return values.get("billing_enabled") == "true" and values.get("printer_kill_switch_enabled") == "true" + + +async def _get_budget_window_start_utc(db: AsyncSession) -> datetime: + result = await db.execute( + select(Settings).where(Settings.key.in_(["finance_budget_reset_day", "finance_budget_reset_timezone"])) + ) + values = {setting.key: setting.value for setting in result.scalars().all()} + + desired_day = 1 + try: + parsed = int(values.get("finance_budget_reset_day") or 1) + if 1 <= parsed <= 31: + desired_day = parsed + except (TypeError, ValueError): + pass + + timezone_name = values.get("finance_budget_reset_timezone") or "UTC" + try: + tz = ZoneInfo(timezone_name) + except ZoneInfoNotFoundError: + tz = ZoneInfo("UTC") + + now = datetime.now(tz) + current_month_reset_day = min(desired_day, calendar.monthrange(now.year, now.month)[1]) + if now.day < current_month_reset_day: + month = now.month - 1 + year = now.year + if month == 0: + month = 12 + year -= 1 + else: + month = now.month + year = now.year + + reset_day = min(desired_day, calendar.monthrange(year, month)[1]) + return datetime(year, month, reset_day, tzinfo=tz).astimezone(timezone.utc) + + +async def _cost_center_spend(db: AsyncSession, cost_center_id: int, *, monthly: bool) -> float: + spend_expr = case((WalletTransaction.amount < 0, -WalletTransaction.amount), else_=0.0) + conditions = [ + WalletTransaction.cost_center_id == cost_center_id, + WalletTransaction.cost_center_id.is_not(None), + ] + if monthly: + conditions.append(WalletTransaction.created_at >= await _get_budget_window_start_utc(db)) + + result = await db.execute(select(func.coalesce(func.sum(spend_expr), 0.0)).where(*conditions)) + return float(result.scalar() or 0.0) + + +async def _cost_center_open_queue_reservations( + db: AsyncSession, + cost_center_id: int, + *, + exclude_queue_item_id: int | None = None, +) -> float: + conditions = [ + PrintQueueItem.cost_center_id == cost_center_id, + PrintQueueItem.status.in_(("pending", "printing")), + ] + if exclude_queue_item_id is not None: + conditions.append(PrintQueueItem.id != exclude_queue_item_id) + + result = await db.execute(select(func.coalesce(func.sum(PrintQueueItem.estimated_cost), 0.0)).where(*conditions)) + return float(result.scalar() or 0.0) + + +async def _cost_center_active_budget_reservations(db: AsyncSession, cost_center_id: int) -> float: + result = await db.execute( + select(func.coalesce(func.sum(BudgetReservation.amount), 0.0)).where( + BudgetReservation.cost_center_id == cost_center_id, + BudgetReservation.status == "active", + ) + ) + return float(result.scalar() or 0.0) + + +async def validate_print_budget( + db: AsyncSession, + *, + cost_center_id: int | None, + estimated_cost: float | None, + current_user: User | None, + quantity: int = 1, + exclude_queue_item_id: int | None = None, +) -> None: + """Validate that a print can be assigned to a cost center budget.""" + if not await is_billing_enabled(db): + return + + if cost_center_id is None: + raise HTTPException(status_code=400, detail="Cost center is required when billing is enabled") + + if estimated_cost is None or estimated_cost <= 0: + raise HTTPException(status_code=400, detail="Estimated cost is required for cost center prints") + + center = await db.scalar(select(CostCenter).where(CostCenter.id == cost_center_id).with_for_update()) + if not center: + raise HTTPException(status_code=404, detail="Cost center not found") + if not center.is_active: + raise HTTPException(status_code=400, detail="Cost center is inactive") + + if current_user is not None and not current_user.is_admin: + if center.is_private: + if center.owner_user_id != current_user.id: + raise HTTPException(status_code=403, detail="You cannot print with this private cost center") + else: + member = await db.scalar( + select(CostCenterMember).where( + CostCenterMember.cost_center_id == cost_center_id, + CostCenterMember.user_id == current_user.id, + ) + ) + if not member or not member.can_print: + raise HTTPException(status_code=403, detail="You cannot print with this cost center") + + budget_limit = center.monthly_budget if center.monthly_budget is not None else center.total_budget + if budget_limit is None: + return + + used = await _cost_center_spend(db, cost_center_id, monthly=center.monthly_budget is not None) + reserved = await _cost_center_open_queue_reservations( + db, + cost_center_id, + exclude_queue_item_id=exclude_queue_item_id, + ) + reserved += await _cost_center_active_budget_reservations(db, cost_center_id) + requested = estimated_cost * max(1, quantity) + available = float(budget_limit) - used - reserved + if requested > available: + raise HTTPException( + status_code=400, + detail=f"Estimated print cost exceeds available cost center budget ({requested:.2f} > {available:.2f})", + ) + + +async def create_budget_reservation( + db: AsyncSession, + *, + cost_center_id: int | None, + estimated_cost: float | None, + current_user: User | None, + source_type: str, + source_id: int | None, + print_archive_id: int | None = None, +) -> BudgetReservation | None: + if not await is_billing_enabled(db): + return None + + if cost_center_id is None: + raise HTTPException(status_code=400, detail="Cost center is required when billing is enabled") + + await validate_print_budget( + db, + cost_center_id=cost_center_id, + estimated_cost=estimated_cost, + current_user=current_user, + ) + reservation = BudgetReservation( + cost_center_id=cost_center_id, + amount=float(estimated_cost or 0.0), + status="active", + source_type=source_type, + source_id=source_id, + print_archive_id=print_archive_id, + ) + db.add(reservation) + await db.flush() + return reservation + + +async def release_budget_reservation( + db: AsyncSession, + *, + source_type: str | None = None, + source_id: int | None = None, + print_archive_id: int | None = None, + status: str = "released", +) -> int: + conditions = [BudgetReservation.status == "active"] + if print_archive_id is not None: + conditions.append(BudgetReservation.print_archive_id == print_archive_id) + else: + conditions.extend( + [ + BudgetReservation.source_type == source_type, + BudgetReservation.source_id == source_id, + ] + ) + + result = await db.execute(select(BudgetReservation).where(*conditions)) + reservations = result.scalars().all() + for reservation in reservations: + reservation.status = status + reservation.released_at = datetime.now(timezone.utc) + if reservations: + await db.flush() + return len(reservations) diff --git a/backend/app/services/finance_defaults.py b/backend/app/services/finance_defaults.py new file mode 100644 index 000000000..2d809ec8b --- /dev/null +++ b/backend/app/services/finance_defaults.py @@ -0,0 +1,68 @@ +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.app.models.finance import CostCenter, CostCenterMember, UserWallet +from backend.app.models.settings import Settings as AppSettingModel +from backend.app.models.user import User +from backend.app.schemas.settings import AppSettings as AppSettingsSchema + + +async def ensure_user_finance_defaults(db: AsyncSession, user: User) -> bool: + """Ensure wallet and private cost center defaults exist for a user. + + Returns True when database objects were created or changed. + """ + changed = False + + wallet = (await db.execute(select(UserWallet).where(UserWallet.user_id == user.id))).scalar_one_or_none() + if wallet is None: + # Respect admin-configured currency if present, otherwise fall back to app default + default_currency = AppSettingsSchema().currency + result = await db.execute(select(AppSettingModel).where(AppSettingModel.key == "currency")) + setting = result.scalar_one_or_none() + currency = setting.value if setting and setting.value else default_currency + db.add(UserWallet(user_id=user.id, balance=0.0, currency=currency)) + changed = True + + private_center = ( + ( + await db.execute( + select(CostCenter) + .where( + CostCenter.is_private.is_(True), + CostCenter.owner_user_id == user.id, + ) + .order_by(CostCenter.id.asc()) + ) + ) + .scalars() + .first() + ) + + if private_center is None: + private_center = CostCenter( + name=user.username, + is_active=True, + is_private=True, + owner_user_id=user.id, + ) + db.add(private_center) + await db.flush() + changed = True + elif private_center.name != user.username: + private_center.name = user.username + changed = True + + membership = ( + await db.execute( + select(CostCenterMember).where( + CostCenterMember.cost_center_id == private_center.id, + CostCenterMember.user_id == user.id, + ) + ) + ).scalar_one_or_none() + if membership is None: + db.add(CostCenterMember(cost_center_id=private_center.id, user_id=user.id, can_print=True)) + changed = True + + return changed diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py index c5ea2f7f4..15f3572f7 100644 --- a/backend/app/services/print_scheduler.py +++ b/backend/app/services/print_scheduler.py @@ -7,6 +7,7 @@ import time from datetime import datetime, timezone from pathlib import Path +from fastapi import HTTPException from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -15,6 +16,7 @@ from backend.app.core.config import settings from backend.app.core.database import async_session, run_with_retry from backend.app.core.tasks import spawn_background_task from backend.app.models.archive import PrintArchive +from backend.app.models.finance import CostCenter, CostCenterMember from backend.app.models.library import LibraryFile from backend.app.models.print_queue import PrintQueueItem from backend.app.models.printer import Printer @@ -22,6 +24,7 @@ from backend.app.models.settings import Settings from backend.app.models.smart_plug import SmartPlug from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment +from backend.app.models.user import User from backend.app.services.bambu_ftp import ( cache_3mf_download, delete_file_async, @@ -30,6 +33,7 @@ from backend.app.services.bambu_ftp import ( with_ftp_retry, ) from backend.app.services.filament_deficit import compute_deficit_for_queue_item +from backend.app.services.finance_budget import validate_print_budget from backend.app.services.notification_service import notification_service from backend.app.services.printer_manager import printer_manager, supports_drying from backend.app.services.smart_plug_manager import smart_plug_manager @@ -2054,6 +2058,44 @@ class PrintScheduler: """ logger.info("Starting queue item %s", item.id) + try: + queue_user = await db.get(User, item.created_by_id) if item.created_by_id is not None else None + if queue_user is not None and item.cost_center_id is not None and not queue_user.is_admin: + center = await db.scalar( + select(CostCenter).where(CostCenter.id == item.cost_center_id).with_for_update() + ) + if not center: + raise HTTPException(status_code=404, detail="Cost center not found") + if not center.is_active: + raise HTTPException(status_code=400, detail="Cost center is inactive") + if center.is_private: + if center.owner_user_id != queue_user.id: + raise HTTPException(status_code=403, detail="You cannot print with this private cost center") + else: + member = await db.scalar( + select(CostCenterMember).where( + CostCenterMember.cost_center_id == item.cost_center_id, + CostCenterMember.user_id == queue_user.id, + ) + ) + if not member or not member.can_print: + raise HTTPException(status_code=403, detail="You cannot print with this cost center") + await validate_print_budget( + db, + cost_center_id=item.cost_center_id, + estimated_cost=item.estimated_cost, + current_user=queue_user, + exclude_queue_item_id=item.id, + ) + except HTTPException as exc: + item.status = "failed" + item.error_message = getattr(exc, "detail", str(exc)) + item.completed_at = datetime.now(timezone.utc) + await db.commit() + logger.error("Queue item %s: Budget check failed: %s", item.id, item.error_message) + await self._power_off_if_needed(db, item) + return + # Get printer first (needed for both paths) result = await db.execute(select(Printer).where(Printer.id == item.printer_id)) printer = result.scalar_one_or_none() @@ -2126,6 +2168,7 @@ class PrintScheduler: original_filename=filename, created_by_id=item.created_by_id, project_id=item.project_id, + cost_center_id=item.cost_center_id, ) if archive: item.archive_id = archive.id @@ -2288,6 +2331,7 @@ class PrintScheduler: archive.id, ams_mapping=ams_mapping, created_by_id=item.created_by_id, + cost_center_id=item.cost_center_id, plate_id=item.plate_id, )