Add auto-detect for transfers

- Auto-detect for transfers between accounts (accounts must be enabled)
- Add response display in get/post requests
- improve tools methods

Author: Royalphax

firefly3.py

@@ -21,18 +21,21 @@ def __init__(self):
         self.token = PERSONAL_TOKEN_DEFAULT
         self.hostname = HOSTNAME_DEFAULT
         self.name_format = ACCOUNTS_NAME_FORMAT_DEFAULT
+        self.auto_detect_transfers = bool(AUTO_DETECT_TRANSFERS_DEFAULT)
+        self.transfer_source_transaction = TRANSFER_SOURCE_TRANSACTION_NAME_DEFAULT.strip().split(",")
+        self.transfer_destination_transaction = TRANSFER_DESTINATION_TRANSACTION_NAME_DEFAULT.strip().split(",")
         self.headers = None
 
     def _post(self, endpoint, payload):
         response = requests.post("{}{}".format(self.hostname, endpoint), json=payload, headers=self.headers)
         if response.status_code != 200:
-            raise ValueError("Request to your Firefly3 instance failed. Please double check your personal token.")
+            raise ValueError("Request to your Firefly3 instance failed. Please double check your personal token. Error : " + str(response.json()))
         return response.json()
 
     def _get(self, endpoint, params=None):
         response = requests.get("{}{}".format(self.hostname, endpoint), params=params, headers=self.headers)
         if response.status_code != 200:
-            raise ValueError("Request to your Firefly3 instance failed. Please double check your personal token.")
+            raise ValueError("Request to your Firefly3 instance failed. Please double check your personal token. Error : " + str(response.json()))
         return response.json()
 
     def validate(self):
@@ -110,14 +113,15 @@ def __init__(self, f3_cli, account_id):
         self.f3_cli = f3_cli
         self.account_id = int(account_id)
         self.payloads = []
-        self.transfer_out = [] # [date, montant]
-        self.transfer_in = []  # [date, montant]
-        # --> On extrait pour chaque compte, une liste unique de dates. Et pour chacunes de ces dates on regarde,
-        # quel compte a emis, quel compte a reçu. Une fois qu'on les a, on crée le transfert (add_transfert?) et pas oublier
-        # de supprimer les 2 transactions du payload (ou alors juste on ne les ajotue pas au payload dans ce genre de cas
+        self.transfer_out = {}
+        self.transfer_in = {}
 
     def __len__(self):
-        return len(self.payloads)
+        count = 0
+        for transfer_list in [self.transfer_in, self.transfer_out]:
+            for date in transfer_list:
+                count = count + len(transfer_list[date])
+        return len(self.payloads) + count
 
     def add_transaction(self, ca_payload):
         payload = {"transactions": [{}]}
@@ -128,22 +132,12 @@ def add_transaction(self, ca_payload):
         renames = get_key_from_value(self.f3_cli.a_rename_transaction, transaction_name)
         transaction["description"] = renames[0] if len(renames) > 0 else transaction_name
 
-        date = time.mktime(time.strptime(ca_payload["dateOperation"], '%b %d, %Y %H:%M:%S %p'))
-        transaction["date"] = time.strftime("%Y-%m-%d", time.gmtime(date))
+        date = time.mktime(time.strptime(ca_payload["dateOperation"], '%b %d, %Y, %H:%M:%S %p'))
+        transaction["date"] = time.strftime("%Y-%m-%dT%T", time.gmtime(date))
 
         transaction["amount"] = abs(ca_payload["montant"])
         transaction["currency_code"] = ca_payload["idDevise"]
 
-        accounts = get_key_from_value(self.f3_cli.aa_account, transaction_name)
-        if ca_payload["montant"] > 0:
-            transaction["type"] = "deposit"
-            transaction["source_name"] = accounts[0] if len(accounts) > 0 else "Cash account"
-            transaction["destination_id"] = self.account_id
-        else:
-            transaction["type"] = "withdrawal"
-            transaction["source_id"] = self.account_id
-            transaction["destination_name"] = accounts[0] if len(accounts) > 0 else "Cash account"
-
         budgets = get_key_from_value(self.f3_cli.aa_budget, transaction_name)
         if len(budgets) != 0:
             transaction["budget_id"] = self.f3_cli.get_budget_id(budgets[0])
@@ -157,12 +151,86 @@ def add_transaction(self, ca_payload):
             tags.append(tag)
         transaction["tags"] = tags
 
-        transaction["notes"] = transaction_name
+        transaction["notes"] = "CREDIT AGRICOLE NAME : " + transaction_name
+
+        if self.f3_cli.auto_detect_transfers and is_in_list(self.f3_cli.transfer_source_transaction, transaction_name):
+            key = transaction["date"]
+            if key not in self.transfer_out:
+                self.transfer_out[key] = []
+            self.transfer_out[key].append(payload)
 
-        if
-        self.payloads.append(payload)
+        elif self.f3_cli.auto_detect_transfers and is_in_list(self.f3_cli.transfer_destination_transaction, transaction_name):
+            key = transaction["date"]
+            if key not in self.transfer_in:
+                self.transfer_in[key] = []
+            self.transfer_in[key].append(payload)
+
+        else:
+            accounts = get_key_from_value(self.f3_cli.aa_account, transaction_name)
+            if ca_payload["montant"] > 0:
+                transaction["type"] = "deposit"
+                transaction["source_name"] = accounts[0] if len(accounts) > 0 else "Cash account"
+                transaction["destination_id"] = self.account_id
+            else:
+                transaction["type"] = "withdrawal"
+                transaction["source_id"] = self.account_id
+                transaction["destination_name"] = accounts[0] if len(accounts) > 0 else "Cash account"
+
+            self.payloads.append(payload)
 
     def post(self):
         for payload in self.payloads:
             print(".", end='')
             self.f3_cli._post(endpoint=_TRANSACTIONS_ENDPOINT, payload=payload)
+
+    @staticmethod
+    def post_transfers(f3transactions_list, f3_cli):
+        payloads = []
+        detected_transfers = 0
+
+        # Loop through all transaction packages
+        for f3_from_transactions in f3transactions_list:
+            # Count detected transfers for later test
+            detected_transfers = detected_transfers + len(f3_from_transactions) - len(f3_from_transactions.payloads)
+            # Loop through dates of outgoing transfers of the above transaction package
+            for date_out in f3_from_transactions.transfer_out.keys():
+                # Loop through outgoing transfers for the above date
+                for transfer_out in f3_from_transactions.transfer_out[date_out]:
+                    # Get the amount
+                    amount_out = transfer_out["transactions"][0]["amount"]
+                    # Now loop through other transaction packages
+                    for f3_to_transactions in f3transactions_list:
+                        # If the above transaction package is the same than the first one, skip  it
+                        if f3_to_transactions.account_id == f3_from_transactions.account_id:
+                            continue
+                        # Loop through dates of incoming transfers
+                        for date_in in f3_to_transactions.transfer_in.keys():
+                            # If the incoming transfer date is the same than our outgoing transfer date, check amounts
+                            if date_in == date_out:
+                                # Loop through incoming transfers
+                                for transfer_in in f3_to_transactions.transfer_in[date_in]:
+                                    # Get the amount
+                                    amount_in = transfer_in["transactions"][0]["amount"]
+                                    # If incoming transfer amount is the same than outgoing transfer amount,
+                                    # it means that we found a corresponding incoming transfer for our outgoing transfer
+                                    if amount_in == amount_out:
+                                        transfer_out["transactions"][0]["type"] = "transfer"
+                                        # We clarify the transfer payload (who's the source and destination)
+                                        transfer_out["transactions"][0]["source_id"] = f3_from_transactions.account_id
+                                        transfer_out["transactions"][0]["destination_id"] = f3_to_transactions.account_id
+                                        # We rename the transaction
+                                        transfer_out["transactions"][0]["description"] = "Personal transfer"
+                                        # We save the payload to push it later
+                                        payloads.append(transfer_out)
+
+        # Check amount of transfers
+        if detected_transfers % 2 != 0 or len(payloads) * 2 != detected_transfers:
+            print("\nWARN: Wrong quantity of transfers detected (" + str(detected_transfers) + ") for " + str(len(payloads)) + " payload(s). You must double check your \"transfer-source-transaction-name\" and \"transfer-destination-transaction-name\" because some transfers hadn't been recognized.")
+
+        # Now push each payload
+        for payload in payloads:
+            print(".", end='')
+            f3_cli._post(endpoint=_TRANSACTIONS_ENDPOINT, payload=payload)
+
+
+

main.py

@@ -68,6 +68,9 @@
         f3_cli.name_format = str(firefly3_section.get(ACCOUNTS_NAME_FORMAT_FIELD, ACCOUNTS_NAME_FORMAT_DEFAULT))
         f3_cli.hostname = str(firefly3_section.get(HOSTNAME_FIELD, HOSTNAME_DEFAULT))
         f3_cli.token = str(firefly3_section.get(PERSONAL_TOKEN_FIELD, PERSONAL_TOKEN_DEFAULT))
+        f3_cli.auto_detect_transfers = bool(settings_section.get(AUTO_DETECT_TRANSFERS_FIELD, AUTO_DETECT_TRANSFERS_DEFAULT))
+        f3_cli.transfer_source_transaction = str(settings_section.get(TRANSFER_SOURCE_TRANSACTION_NAME_FIELD, TRANSFER_SOURCE_TRANSACTION_NAME_DEFAULT)).strip().split(",")
+        f3_cli.transfer_destination_transaction = str(settings_section.get(TRANSFER_DESTINATION_TRANSACTION_NAME_FIELD, TRANSFER_DESTINATION_TRANSACTION_NAME_DEFAULT)).strip().split(",")
         f3_cli.init_auto_assign_values(a_rename_transaction_section, aa_budget_section, aa_category_section, aa_account_section, aa_tags_section)
         f3_cli.validate()
 
@@ -78,6 +81,9 @@
         f3_accounts = f3_cli.get_accounts(account_type="asset")
         f3_accounts_number = [account.get("attributes").get("account_number") for account in f3_accounts]
 
+        # Keep track of transactions for each account
+        account_transactions = []
+
         # Loop through existing CreditAgricole accounts declared in config file
         for account in ca_cli.get_accounts():
 
@@ -115,3 +121,10 @@
                 # Push transactions to Firefly3
                 transactions.post()
                 print(" Done!")
+
+            account_transactions.append(transactions)
+
+        if f3_cli.auto_detect_transfers:
+            print("-> Pushing transfers to Firefly3 instance ", end='')
+            Firefly3Transactions.post_transfers(account_transactions, f3_cli)
+            print(" Done!")

tool.py

@@ -1,23 +1,44 @@
-def get_key_from_value(dict, str_to_find):
+def get_key_from_value(dict, str_to_check):
     """
     :param dict: dictionnary dict[key] = values
-    :param str_to_find: occurence to find in the values for each key (* can be used)
+    :param str_to_check: occurence to find in the values for each key (* can be used)
     :return: the key where occurence is found in one of the values
     """
     output = []
-    str_to_find = str_to_find.upper()
     for key in dict.keys():
         for value in dict[key]:
-            value = value.upper()
-            add_key = False
-            if value.startswith("*") and value.endswith("*") and value[1:-1] in str_to_find:
-                add_key = True
-            elif value.startswith("*") and str_to_find.endswith(value[1:]):
-                add_key = True
-            elif value.endswith("*") and str_to_find.startswith(value[:-1]):
-                add_key = True
-            elif value == str_to_find:
-                add_key = True
-            if add_key and key not in output:
+            if is_equal(value, str_to_check) and key not in output:
                 output.append(key.title().replace("_", " "))
     return output
+
+
+def is_in_list(list_to_check, str_to_check):
+    """
+    :param value: Value to find in str_to_check
+    :param list_to_check: List of string to check
+    :return: if value is found in str_to_check
+    """
+    for elem in list_to_check:
+        if is_equal(elem, str_to_check):
+            return True
+    return False
+
+
+def is_equal(value, str_to_check):
+    """
+    :param value: Value to find in str_to_check
+    :param str_to_check: String in which we check
+    :return: if value is found in str_to_check using star indicators
+    """
+    output = False
+    value = value.upper()
+    str_to_check = str_to_check.upper()
+    if value.startswith("*") and value.endswith("*") and value[1:-1] in str_to_check:
+        output = True
+    elif value.startswith("*") and str_to_check.endswith(value[1:]):
+        output = True
+    elif value.endswith("*") and str_to_check.startswith(value[:-1]):
+        output = True
+    elif value == str_to_check:
+        output = True
+    return output