dusker
8877de2cef
See https://de.wikipedia.org/wiki/EPC-QR-Code#EPC-QR-Code_f%C3%BCr_%C3%9Cberweisung_erstellen for more information about the EPC coding Co-authored-by: dusker <dusker@gmx.de> Reviewed-on: #61 Co-authored-by: dusker <jens.graef+ezgg@posteo.de> Co-committed-by: dusker <jens.graef+ezgg@posteo.de>
105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
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()
|