import io import logging import qrcode from collections.abc import Callable from datetime import datetime from decimal import Decimal, ROUND_DOWN from typing import Optional from src.ezgg_lan_manager.services.DatabaseService import DatabaseService from src.ezgg_lan_manager.types.Transaction import Transaction logger = logging.getLogger(__name__.split(".")[-1]) class InsufficientFundsError(Exception): pass class AccountingService: def __init__(self, db_service: DatabaseService) -> None: self._db_service = db_service self._update_hooks: set[Callable] = set() def add_update_hook(self, update_hook: Callable) -> None: """ Adds a function to this service, which is called whenever the account balance changes """ self._update_hooks.add(update_hook) async def add_balance(self, user_id: int, balance_to_add: Decimal, reference: str) -> Decimal: await self._db_service.add_transaction(Transaction( user_id=user_id, value=balance_to_add, is_debit=False, reference=reference, transaction_date=datetime.now() )) logger.debug(f"Added balance of {self.make_euro_string_from_decimal(balance_to_add)} to user with ID {user_id}") for update_hook in self._update_hooks: await update_hook() return await self.get_balance(user_id) async def remove_balance(self, user_id: int, balance_to_remove: Decimal, reference: str) -> Decimal: current_balance = await self.get_balance(user_id) if (current_balance - balance_to_remove) < 0: raise InsufficientFundsError await self._db_service.add_transaction(Transaction( user_id=user_id, value=balance_to_remove, is_debit=True, reference=reference, transaction_date=datetime.now() )) logger.debug( f"Removed balance of {self.make_euro_string_from_decimal(balance_to_remove)} to user with ID {user_id}") for update_hook in self._update_hooks: await update_hook() return await self.get_balance(user_id) async def get_balance(self, user_id: int) -> Decimal: balance_buffer = Decimal("0") for transaction in await self._db_service.get_all_transactions_for_user(user_id): if transaction.is_debit: balance_buffer -= transaction.value else: balance_buffer += transaction.value return balance_buffer async def get_transaction_history(self, user_id: int) -> list[Transaction]: return await self._db_service.get_all_transactions_for_user(user_id) @staticmethod def make_euro_string_from_decimal(euros: Optional[Decimal]) -> str: """ Internally, all money values are euros as decimal. Only when showing them to the user we generate a string. """ if euros is None: return "0.00 €" rounded_decimal = str(euros.quantize(Decimal(".01"), rounding=ROUND_DOWN)) return f"{rounded_decimal} €" @staticmethod def make_payment_qr_image(beneficiary_name, beneficiary_bic, beneficiary_iban, text, amount_euros=None) -> bytes: text = text.replace("\n",";") amount_formatted = "EUR{:.2f}".format(amount_euros) if amount_euros else "" epc_text = f"""BCD 002 1 SCT {beneficiary_bic} {beneficiary_name} {beneficiary_iban} {amount_formatted}, {text} """ qr = qrcode.QRCode( version=6, error_correction=qrcode.constants.ERROR_CORRECT_M, ) qr.add_data(epc_text) img = qr.make_image() img_bytes = io.BytesIO() img.save(img_bytes) return img_bytes.getvalue()