diff --git a/dkb_robo/dkb_robo.py b/dkb_robo/dkb_robo.py index aabbeed..b7f5778 100644 --- a/dkb_robo/dkb_robo.py +++ b/dkb_robo/dkb_robo.py @@ -32,8 +32,9 @@ class DKBRobo(object): chip_tan = False logger = None wrapper = None + dkb_raw = False - def __init__(self, dkb_user=None, dkb_password=None, tan_insert=False, legacy_login=False, debug=False, mfa_device=None, chip_tan=False): + def __init__(self, dkb_user=None, dkb_password=None, tan_insert=False, legacy_login=False, debug=False, mfa_device=None, chip_tan=False, dkb_raw=False): self.dkb_user = dkb_user self.dkb_password = dkb_password self.chip_tan = chip_tan @@ -41,6 +42,7 @@ def __init__(self, dkb_user=None, dkb_password=None, tan_insert=False, legacy_lo self.legacy_login = legacy_login self.logger = logger_setup(debug) self.mfa_device = mfa_device + self.dkb_raw = dkb_raw def __enter__(self): """ Makes DKBRobo a Context Manager """ @@ -94,7 +96,7 @@ def get_points(self): def get_standing_orders(self, uid=None): """ get standing orders """ self.logger.debug('DKBRobo.get_standing_orders()\n') - standingorder = StandingOrder(client=self.wrapper.client) + standingorder = StandingOrder(client=self.wrapper.client, dkb_raw=self.dkb_raw) return standingorder.fetch(uid) def get_transactions(self, transaction_url, atype, date_from, date_to, transaction_type='booked'): diff --git a/dkb_robo/standingorder.py b/dkb_robo/standingorder.py index 68ec7c1..6c9918f 100644 --- a/dkb_robo/standingorder.py +++ b/dkb_robo/standingorder.py @@ -1,18 +1,71 @@ """ Module for handling dkb standing orders """ -from typing import Dict, List +from typing import Dict, List, Optional +from dataclasses import dataclass, field import logging import requests -from dkb_robo.utilities import DKBRoboError +from dkb_robo.utilities import DKBRoboError, Amount, filter_unexpected_fields logger = logging.getLogger(__name__) +@filter_unexpected_fields +@dataclass +class CreditorAccount: + """ class for a single creditor account """ + iban: str = None + bic: str = None + name: str = None + + +@filter_unexpected_fields +@dataclass +class DebtorAccount: + """ class for a single debitor account """ + iban: str = None + accountId: str = None # pylint: disable=C0103 # NOSONAR + + +@filter_unexpected_fields +@dataclass +class Recurrence: + """ class for frequency account """ + frm: str = None + frequency: str = None + holidayExecutionStrategy: str = None # pylint: disable=C0103 # NOSONAR + nextExecutionAt: str = None # pylint: disable=C0103 # NOSONAR + until: str = None + + +@filter_unexpected_fields +@dataclass +class StandingOrderItem: + """ class for a single standing order """ + amount: Optional[Amount] = None + creditor: Optional[CreditorAccount] = None + debtor: Optional[DebtorAccount] = None + description: str = None + messages: List[str] = field(default_factory=list) + recurrence: Optional[Recurrence] = None + status: str = None + + def __post_init__(self): + self.amount = Amount(**self.amount) + self.creditor['creditorAccount']['name'] = self.creditor.get('name', {}) + self.creditor = CreditorAccount(**self.creditor['creditorAccount']) + if self.debtor and 'debtorAccount' in self.debtor: + self.debtor = DebtorAccount(**self.debtor['debtorAccount']) + # rewrite from - field to frm + self.recurrence['frm'] = self.recurrence.get('from', None) + self.recurrence = Recurrence(**self.recurrence) + + class StandingOrder: """ StandingOrder class """ - def __init__(self, client: requests.Session, base_url: str = 'https://banking.dkb.de/api'): + def __init__(self, client: requests.Session, dkb_raw: bool = False, base_url: str = 'https://banking.dkb.de/api'): self.client = client self.base_url = base_url + self.dkb_raw = dkb_raw self.uid = None def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]: @@ -23,20 +76,28 @@ def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]: if 'data' in full_list: for ele in full_list['data']: - try: - amount = float(ele.get('attributes', {}).get('amount', {}).get('value', None)) - except Exception as err: - logger.error('amount conversion error: %s', err) - amount = None - - _tmp_dic = { - 'amount': amount, - 'currencycode': ele.get('attributes', {}).get('amount', {}).get('currencyCode', None), - 'purpose': ele.get('attributes', {}).get('description', None), - 'recpipient': ele.get('attributes', {}).get('creditor', {}).get('name', None), - 'creditoraccount': ele.get('attributes', {}).get('creditor', {}).get('creditorAccount', None), - 'interval': ele.get('attributes', {}).get('recurrence', None)} - so_list.append(_tmp_dic) + standingorder_obj = StandingOrderItem(**ele['attributes']) + + if self.dkb_raw: + so_list.append(standingorder_obj) + else: + so_list.append({ + 'amount': standingorder_obj.amount.value, + 'currencycode': standingorder_obj.amount.currencyCode, + 'purpose': standingorder_obj.description, + 'recipient': standingorder_obj.creditor.name, + 'creditoraccount': { + 'iban': standingorder_obj.creditor.iban, + 'bic': standingorder_obj.creditor.bic + }, + 'interval': { + 'frequency': standingorder_obj.recurrence.frequency, + 'from': standingorder_obj.recurrence.frm, + 'holidayExecutionStrategy': standingorder_obj.recurrence.holidayExecutionStrategy, + 'nextExecutionAt': standingorder_obj.recurrence.nextExecutionAt, + 'until': standingorder_obj.recurrence.until + } + }) logger.debug('StandingOrder._filter() ended with: %s entries.', len(so_list)) return so_list diff --git a/dkb_robo/utilities.py b/dkb_robo/utilities.py index 1933906..8c724a9 100644 --- a/dkb_robo/utilities.py +++ b/dkb_robo/utilities.py @@ -6,11 +6,14 @@ from string import digits, ascii_letters from typing import List, Tuple from datetime import datetime, timezone -from dataclasses import fields +from dataclasses import dataclass, fields import time import re +logger = logging.getLogger(__name__) + + def get_dateformat(): """ get date format """ return '%d.%m.%Y', '%Y-%m-%d' @@ -19,6 +22,19 @@ def get_dateformat(): LEGACY_DATE_FORMAT, API_DATE_FORMAT = get_dateformat() JSON_CONTENT_TYPE = 'application/vnd.api+json' +@dataclass +class Amount: + """ Amount data class, roughly based on the JSON API response. """ + value: float = None + currencyCode: str = None + + def __post_init__(self): + # convert value to float + try: + self.value = float(self.value) + except Exception as err: + logger.error('Account.__post_init: conversion error: %s', err) + self.value = None class DKBRoboError(Exception): """ dkb-robo exception class """ diff --git a/test/test_standingorder.py b/test/test_standingorder.py index 3fbb4e9..3213d86 100644 --- a/test/test_standingorder.py +++ b/test/test_standingorder.py @@ -32,6 +32,7 @@ def setUp(self, mock_session): self.dir_path = os.path.dirname(os.path.realpath(__file__)) self.logger = logging.getLogger('dkb_robo') self.dkb = StandingOrder(client=mock_session) + self.maxDiff = None @patch('dkb_robo.standingorder.StandingOrder._filter') def test_001_fetch(self, mock_filter): @@ -95,13 +96,14 @@ def test_005__filter(self): } } }]} - result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description', 'recpipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01'}}] + result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description', 'recipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01', 'holidayExecutionStrategy': None}}] self.assertEqual(result, self.dkb._filter(full_list)) def test_006__filter(self): """ test StandingOrder._filter() with list from file """ so_list = json_load(self.dir_path + '/mocks/so.json') - result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description1', 'recpipient': 'name1', 'creditoraccount': {'iban': 'iban1', 'bic': 'bic1'}, 'interval': {'from': '2022-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-01'}}, {'amount': 200.0, 'currencycode': 'EUR', 'purpose': 'description2', 'recpipient': 'name2', 'creditoraccount': {'iban': 'iban2', 'bic': 'bic2'}, 'interval': {'from': '2022-02-01', 'until': '2025-12-02', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-02'}}, {'amount': 300.0, 'currencycode': 'EUR', 'purpose': 'description3', 'recpipient': 'name3', 'creditoraccount': {'iban': 'iban3', 'bic': 'bic3'}, 'interval': {'from': '2022-03-01', 'until': '2025-03-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-03-01'}}] + self.dkb.dkb_raw = False + result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description1', 'recipient': 'name1', 'creditoraccount': {'iban': 'iban1', 'bic': 'bic1'}, 'interval': {'from': '2022-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-01'}}, {'amount': 200.0, 'currencycode': 'EUR', 'purpose': 'description2', 'recipient': 'name2', 'creditoraccount': {'iban': 'iban2', 'bic': 'bic2'}, 'interval': {'from': '2022-02-01', 'until': '2025-12-02', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-02'}}, {'amount': 300.0, 'currencycode': 'EUR', 'purpose': 'description3', 'recipient': 'name3', 'creditoraccount': {'iban': 'iban3', 'bic': 'bic3'}, 'interval': {'from': '2022-03-01', 'until': '2025-03-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-03-01'}}] self.assertEqual(result, self.dkb._filter(so_list)) def test_007__filter(self): @@ -129,10 +131,10 @@ def test_007__filter(self): } } }]} - result = [{'amount': None, 'currencycode': None, 'purpose': 'description', 'recpipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01'}}] + result = [{'amount': None, 'currencycode': None, 'purpose': 'description', 'recipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01', 'holidayExecutionStrategy': None}}] with self.assertLogs('dkb_robo', level='INFO') as lcm: self.assertEqual(result, self.dkb._filter(full_list)) - self.assertIn("ERROR:dkb_robo.standingorder:amount conversion error: could not convert string to float: 'aa'", lcm.output) + self.assertIn("ERROR:dkb_robo.utilities:Account.__post_init: conversion error: could not convert string to float: 'aa'", lcm.output) if __name__ == '__main__':