initial commit

2026-05-20 22:51:52 +02:00
parent 45ad5f164a
commit 85619feed5
43 changed files with 2592 additions and 0 deletions
@@ -0,0 +1,150 @@
+import io
+import logging
+from decimal import Decimal, ROUND_DOWN
+from typing import Optional
+
+import httpx
+import qrcode
+
+from elm.types import Transaction, User, PayPalConfiguration
+
+logger = logging.getLogger(__name__.split(".")[-1])
+
+
+class InsufficientFundsError(Exception):
+    pass
+
+
+class AccountingService:
+    def __init__(self, paypal_config: PayPalConfiguration) -> None:
+        self._paypal_config = paypal_config
+
+    async def get_paypal_access_token(self) -> str:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                "https://api-m.sandbox.paypal.com/v1/oauth2/token",
+                auth=(
+                    self._paypal_config.client_id_sandbox,
+                    self._paypal_config.secret_sandbox,
+                ),
+                headers={
+                    "Content-Type": "application/x-www-form-urlencoded",
+                },
+                data={
+                    "grant_type": "client_credentials"
+                }
+            )
+
+            response.raise_for_status()
+
+            data = response.json()
+            return data["access_token"]
+
+    async def start_paypal_process(self, user_name: str, amount: Decimal) -> str:
+        async with httpx.AsyncClient() as client:
+            access_token = await self.get_paypal_access_token()
+            response = await client.post(
+                url="https://api-m.sandbox.paypal.com/v2/checkout/orders/",
+                headers={
+                    "Authorization": f"Bearer {access_token}"
+                },
+                json={
+                    "intent": "CAPTURE",
+                    "purchase_units": [
+                        {
+                            "custom_id": user_name,
+                            "amount": {
+                                "currency_code": "EUR",
+                                "value": str(amount)
+                            }
+                        }
+                    ]
+                }
+            )
+            approval_url = next(
+                link["href"]
+                for link in response.json()["links"]
+                if link["rel"] == "approve"
+            )
+            return approval_url
+
+
+    async def add_balance(self, user_name: str, balance_to_add: Decimal, title: str) -> Decimal:
+        user = await User.find_one(User.user_name == user_name)
+        if not user:
+            raise KeyError("User does not exist")
+        await Transaction(
+            user_name=user_name,
+            value=balance_to_add,
+            is_debit=False,
+            title=title
+        ).save()
+        logger.debug(f"Added balance of {self.make_euro_string_from_decimal(balance_to_add)} to user '{user_name}'")
+        return await self.get_balance(user_name)
+
+    async def remove_balance(self, user_name: str, balance_to_remove: Decimal, title: str) -> Decimal:
+        current_balance = await self.get_balance(user_name)
+        if (current_balance - balance_to_remove) < 0:
+            raise InsufficientFundsError
+
+        await Transaction(
+            user_name=user_name,
+            value=balance_to_remove,
+            is_debit=True,
+            title=title
+        ).save()
+        logger.debug(
+            f"Removed balance of {self.make_euro_string_from_decimal(balance_to_remove)} from user '{user_name}'")
+        return await self.get_balance(user_name)
+
+    async def get_balance(self, user_name: str) -> Decimal:
+        balance_buffer = Decimal("0")
+        for transaction in await self.get_transaction_history(user_name):
+            if transaction.is_debit:
+                balance_buffer -= transaction.value
+            else:
+                balance_buffer += transaction.value
+        return balance_buffer
+
+    @staticmethod
+    async def get_transaction_history(user_name: str) -> list[Transaction]:
+        user = await User.find_one(User.user_name == user_name)
+        if not user:
+            raise KeyError("User does not exist")
+        return await Transaction.find_many(Transaction.user_name == user_name).to_list()
+
+    @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()