Cora OFX Importer refactoring

Move OFX importer for Cora checking account to its own module.

Author: rennerocha

thebook/bookkeeping/importers/__init__.py

@@ -1,11 +1,13 @@
 import logging
 
+from django.conf import settings
 from django.utils.translation import gettext as _
 
 from thebook.bookkeeping.importers.csv import CSVImporter
 from thebook.bookkeeping.importers.ofx import OFXImporter
-from thebook.bookkeeping.models import Transaction
+from thebook.bookkeeping.models import BankAccount, Transaction
 from thebook.integrations.cora.credit_card_invoice import CoraCreditCardInvoiceImporter
+from thebook.integrations.cora.ofx_importer import CoraOFXImporter
 
 logger = logging.getLogger(__name__)
 
@@ -23,12 +25,21 @@ def __init__(self, message=None):
 def import_transactions(
     transactions_file, file_type, bank_account, user, start_date, end_date
 ):
-    importers = {
-        "csv": CSVImporter,
-        "csv_cora_credit_card": CoraCreditCardInvoiceImporter,
-        "ofx": OFXImporter,
-    }
-    importer = importers.get(file_type) or None
+    cora_bank_account, _ = BankAccount.objects.get_or_create(
+        name=settings.CORA_BANK_ACCOUNT
+    )
+
+    importer = None
+    if file_type == "csv":
+        importer = CSVImporter
+    elif file_type == "csv_cora_credit_card":
+        importer = CoraCreditCardInvoiceImporter
+    elif file_type == "ofx":
+        if bank_account == cora_bank_account:
+            importer = CoraOFXImporter
+        else:
+            importer = OFXImporter
+
     if importer is None:
         raise ImportTransactionsError(
             _("Unable to find a suitable file importer for this file.")
@@ -46,6 +57,17 @@ def import_transactions(
                 update_fields=["description", "amount"],
                 unique_fields=["reference"],
             )
+        elif file_type == "ofx" and bank_account == cora_bank_account:
+            ofx_importer = importer(transactions_file)
+            transactions = ofx_importer.get_transactions(
+                start_date, end_date, exclude_existing=True
+            )
+            Transaction.objects.bulk_create(
+                transactions,
+                update_conflicts=True,
+                update_fields=["description", "amount"],
+                unique_fields=["reference"],
+            )
         else:
             importer(transactions_file, bank_account, user).run(start_date, end_date)
     except Exception as err:

thebook/bookkeeping/tests/importers/test_ofx.py

@@ -28,25 +28,6 @@ def test_raise_error_when_providing_invalid_ofx_file(db, bank_account, user):
         ofx_importer = OFXImporter(invalid_ofx_file, bank_account, user)
 
 
-def test_cora_ofx_file_with_one_transaction(db, request, bank_account, user):
-    ofx_file_path = request.path.parent / "data" / "cora-one-transaction.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-
-        transactions = ofx_importer.run()
-
-        assert len(transactions) == 1
-        transaction = transactions[0]
-        assert transaction.reference == "16b3dab5-d1ca-41e1-87c2-a26920ae70ac"
-        assert transaction.date == datetime.date(2024, 8, 19)
-        assert transaction.description == "Debito em Conta"
-        assert transaction.amount == decimal.Decimal("-2500.50")
-        assert transaction.notes == ""
-        assert transaction.bank_account == bank_account
-        assert transaction.created_by == user
-        assert transaction.category is None
-
-
 def test_bradesco_ofx_file_with_one_transaction(db, request, bank_account, user):
     ofx_file_path = request.path.parent / "data" / "bradesco-one-transaction.ofx"
     with open(ofx_file_path, "r") as ofx_file:
@@ -66,29 +47,6 @@ def test_bradesco_ofx_file_with_one_transaction(db, request, bank_account, user)
         assert transaction.category is None
 
 
-def test_ofx_file_with_multiple_transactions(db, request, bank_account, user):
-    ofx_file_path = request.path.parent / "data" / "cora-multiple-transactions.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-
-        transactions = ofx_importer.run()
-
-        references = sorted([transaction.reference for transaction in transactions])
-        expected_references = sorted(
-            [
-                "3825c888-1017-497d-bee2-c0737d5dafc0",
-                "63c41497-5e0b-4a3c-ada1-d9abebb0af39",
-                "69b66fe5-0360-466c-b2e5-08efc1768389",
-                "39221b44-f648-44ec-b6d1-1d8eefe54e02",
-                "6b01e19f-fc2f-4ea7-b524-6f6c634aea63",
-            ]
-        )
-
-        assert len(transactions) == 5
-        assert Transaction.objects.count() == 5
-        assert references == expected_references
-
-
 def test_bradesco_ofx_file_with_multiple_transactions(db, request, bank_account, user):
     ofx_file_path = request.path.parent / "data" / "bradesco-multiple-transactions.ofx"
     with open(ofx_file_path, "r") as ofx_file:
@@ -100,122 +58,6 @@ def test_bradesco_ofx_file_with_multiple_transactions(db, request, bank_account,
         assert Transaction.objects.count() == 57
 
 
-def test_import_same_ofx_file_twice_will_not_duplicate_transactions(
-    db, request, bank_account, user
-):
-    ofx_file_path = request.path.parent / "data" / "cora-multiple-transactions.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-        transactions = ofx_importer.run()
-        assert Transaction.objects.all().count() == 5
-
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-        transactions = ofx_importer.run()
-        assert Transaction.objects.all().count() == 5
-
-
-def test_import_ofx_file_that_overlaps_other_already_imported_do_not_duplicate_transactions(
-    db, request, bank_account, user
-):
-    ofx_file_path = request.path.parent / "data" / "cora-multiple-transactions.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-        _ = ofx_importer.run()
-        assert Transaction.objects.all().count() == 5
-
-    ofx_file_path = (
-        request.path.parent / "data" / "cora-multiple-transactions-extended.ofx"
-    )
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-        transactions = ofx_importer.run()
-    assert Transaction.objects.all().count() == 7
-
-    references = sorted([transaction.reference for transaction in transactions])
-    expected_references = sorted(
-        [
-            "3825c888-1017-497d-bee2-c0737d5dafc0",
-            "63c41497-5e0b-4a3c-ada1-d9abebb0af39",
-            "69b66fe5-0360-466c-b2e5-08efc1768389",
-            "39221b44-f648-44ec-b6d1-1d8eefe54e02",
-            "6b01e19f-fc2f-4ea7-b524-6f6c634aea63",
-            "b3950c28-8b64-49ed-bdfb-d83a4df39c96",
-            "f5098c7d-81f0-44d9-b0f7-3b8eadee5c34",
-        ]
-    )
-
-    assert len(transactions) == 7
-    assert references == expected_references
-
-
-def test_ofx_file_with_multiple_transactions_filter_by_start_date(
-    db, request, bank_account, user
-):
-    ofx_file_path = request.path.parent / "data" / "cora-multiple-transactions.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-
-        transactions = ofx_importer.run(start_date=datetime.date(2024, 8, 25))
-
-        references = sorted([transaction.reference for transaction in transactions])
-        expected_references = sorted(
-            [
-                "69b66fe5-0360-466c-b2e5-08efc1768389",
-                "39221b44-f648-44ec-b6d1-1d8eefe54e02",
-                "6b01e19f-fc2f-4ea7-b524-6f6c634aea63",
-            ]
-        )
-
-        assert len(transactions) == 3
-        assert references == expected_references
-
-
-def test_ofx_file_with_multiple_transactions_filter_by_end_date(
-    db, request, bank_account, user
-):
-    ofx_file_path = request.path.parent / "data" / "cora-multiple-transactions.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-
-        transactions = ofx_importer.run(end_date=datetime.date(2024, 8, 25))
-
-        references = sorted([transaction.reference for transaction in transactions])
-        expected_references = sorted(
-            [
-                "3825c888-1017-497d-bee2-c0737d5dafc0",
-                "63c41497-5e0b-4a3c-ada1-d9abebb0af39",
-            ]
-        )
-
-        assert len(transactions) == 2
-        assert references == expected_references
-
-
-def test_ofx_file_with_multiple_transactions_filter_by_start_date_and_end_date(
-    db, request, bank_account, user
-):
-    ofx_file_path = request.path.parent / "data" / "cora-multiple-transactions.ofx"
-    with open(ofx_file_path, "r") as ofx_file:
-        ofx_importer = OFXImporter(ofx_file, bank_account, user)
-
-        transactions = ofx_importer.run(
-            start_date=datetime.date(2024, 8, 24), end_date=datetime.date(2024, 9, 15)
-        )
-
-        references = sorted([transaction.reference for transaction in transactions])
-        expected_references = sorted(
-            [
-                "63c41497-5e0b-4a3c-ada1-d9abebb0af39",
-                "69b66fe5-0360-466c-b2e5-08efc1768389",
-                "39221b44-f648-44ec-b6d1-1d8eefe54e02",
-            ]
-        )
-
-        assert len(transactions) == 3
-        assert references == expected_references
-
-
 @pytest.mark.parametrize(
     "ignored_memos,expected_references",
     [

thebook/integrations/cora/ofx_importer.py

@@ -0,0 +1,123 @@
+import csv
+import datetime
+import decimal
+import io
+import uuid
+
+import structlog
+from ofxparse import OfxParser
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+
+from thebook.bookkeeping.models import BankAccount, Category, Transaction
+
+logger = structlog.get_logger(__name__)
+
+
+class InvalidCoraOFXFile(Exception): ...
+
+
+class CoraOFXImporter:
+    def __init__(self, ofx_file):
+        try:
+            self.ofx_parser = OfxParser.parse(ofx_file)
+        except (UnicodeDecodeError, TypeError, ValueError) as exc:
+            logger.error(
+                "CoraOFXImporter.__init__.invalid_cora_ofx_file",
+                start_date=start_date,
+                end_date=end_date,
+                exclude_existing=exclude_existing,
+            )
+            raise InvalidCoraOFXFile() from exc
+
+        self.cora_bank_account, _ = BankAccount.objects.get_or_create(
+            name=settings.CORA_BANK_ACCOUNT
+        )
+        self.cora_credit_card_bank_account, _ = BankAccount.objects.get_or_create(
+            name=settings.CORA_CREDIT_CARD_BANK_ACCOUNT
+        )
+        self.bank_account_transfer_category, _ = Category.objects.get_or_create(
+            name="Transferência entre contas bancárias"
+        )
+        self.user = get_user_model().objects.get_or_create_automation_user()
+
+    def _within_date_range(self, transaction_date, start_date, end_date):
+        date_rules = []
+        if start_date is not None:
+            date_rules.append(transaction_date >= start_date)
+        if end_date is not None:
+            date_rules.append(transaction_date <= end_date)
+        return all(date_rules)
+
+    def get_transactions(
+        self, start_date=None, end_date=None, exclude_existing: bool = True
+    ):
+        logger.info(
+            "CoraOFXImporter.get_transactions.start",
+            start_date=start_date,
+            end_date=end_date,
+            exclude_existing=exclude_existing,
+        )
+
+        transactions = []
+
+        ofx_transactions = self.ofx_parser.account.statement.transactions
+        for transaction in ofx_transactions:
+            if (
+                exclude_existing
+                and Transaction.objects.filter(reference=transaction.id).exists()
+            ):
+                continue
+
+            transaction_date = transaction.date.date()
+            if not self._within_date_range(transaction_date, start_date, end_date):
+                continue
+
+            transaction_notes = ""
+            transaction_category = None
+
+            if "Pagamento da fatura" in transaction.memo:
+                # Cora Credit Card invoice doesn't provide the information when the invoice
+                # is paid, so we need to add it based on the payment that is provided in
+                # the checking account OFX report
+                bank_account_transfer_reference = uuid.uuid4()
+
+                transaction_notes = notes = (
+                    f"Transferência entre contas bancárias - {bank_account_transfer_reference}"
+                )
+                transaction_category = self.bank_account_transfer_category
+
+                transactions.append(
+                    Transaction(
+                        reference=bank_account_transfer_reference,
+                        date=transaction_date,
+                        description=transaction.memo,
+                        amount=-1 * transaction.amount,
+                        notes=f"Transferência entre contas bancárias - {transaction.id}",
+                        bank_account=self.cora_credit_card_bank_account,
+                        category=transaction_category,
+                        source="cora-ofx-importer",
+                        created_by=self.user,
+                    )
+                )
+
+            transactions.append(
+                Transaction(
+                    reference=transaction.id,
+                    date=transaction_date,
+                    description=transaction.memo,
+                    notes=transaction_notes,
+                    amount=transaction.amount,
+                    bank_account=self.cora_bank_account,
+                    category=transaction_category,
+                    source="cora-ofx-importer",
+                    created_by=self.user,
+                )
+            )
+
+        logger.info(
+            "CoraOFXImporter.get_transactions.completed",
+            num_transactions=len(transactions),
+        )
+        return transactions