pluto/handler/sinks/__init__.py

# Other library imports
# import requests
# from json import dumps

# Project imports
# from settings import settings
import sinks.fidor
import sinks.nordigen
import sinks.truelayer
import util
from db import r


class Sinks(util.Base):
    """
    Class to manage calls to various sinks.
    """

    def __init__(self):
        super().__init__()
        self.account_info = {}

    def __irc_started__(self):
        self.startup()

    def startup(self):
        """
        We NEED the other libraries, and we initialise fast, so don't make
        any race conditions by relying on something that might not be there.
        """
        self.fidor = sinks.fidor.Fidor()
        self.nordigen = sinks.nordigen.Nordigen()
        self.truelayer = sinks.truelayer.TrueLayer(self)
        # setattr(self.truelayer, "sinks", self)

    def got_transactions(self, bank, account_id, transactions):
        if not transactions:
            return False
        transaction_ids = [x["transaction_id"] for x in transactions]
        new_key_name = f"new.transactions.{bank}.{account_id}"
        old_key_name = f"transactions.{bank}.{account_id}"
        # for transaction_id in transaction_ids:
        if not transaction_ids:
            return
        r.sadd(new_key_name, *transaction_ids)

        difference = list(r.sdiff(new_key_name, old_key_name))

        difference = util.convert(difference)

        new_transactions = [x for x in transactions if x["transaction_id"] in difference]

        # Rename the new key to the old key so we can run the diff again
        r.rename(new_key_name, old_key_name)
        for transaction in new_transactions:
            self.tx.transaction(transaction)

    def got_account_info(self, subclass, account_infos):
        """
        Called when we get account information from an API provider.
        :param subclass: class name that called it, truelayer, fidor, etc
        :param account_infos: dict of dicts of account information
        :param account_infos: dict
        """

        for bank, accounts in account_infos.items():
            for account in list(accounts):
                if len(account["account_number"]) == 1:
                    account_infos[bank].remove(account)
                    self.log.warning(f"Potentially useless bank account: {account}")
        currencies = [account["currency"] for bank, accounts in account_infos.items() for account in accounts]

        self.account_info = account_infos
        self.currencies = currencies

        # parsed_details =
        # {"EUR": {"IBAN": "xxx", "BIC": "xxx"},
        #  "GBP": {"SORT": "04-04-04", "ACCOUNT": "1922-2993"}}
        # self.markets.distribute_account_details(currencies, account_infos)