fix: Installment transactions import

Author: andreroggeri

.github/workflows/python-package.yml

@@ -34,6 +34,11 @@ jobs:
         run: |
           # stop the build if there are Python syntax errors or undefined names
           flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+      - name: Setup locale # Required for pytest to run with pt_BR.UTF-8
+        run: |
+          sudo apt-get install -y locales
+          sudo locale-gen pt_BR.UTF-8
+          sudo update-locale LANG=pt_BR.UTF-8
       - name: Test with pytest
         run: |
           pytest --cov=brbanks2ynab

brbanks2ynab/cli.py

@@ -17,40 +17,43 @@ def _default_config_path():
 def main():
     logging.basicConfig()
     logger = logging.getLogger('brbanks2ynab')
-
+    
     parser = ArgumentParser(description='Importador de transações de bancos brasileiros para o YNAB')
     parser.add_argument('--debug', action='store_true')
-
+    
     subparsers = parser.add_subparsers(dest='cmd')
-
+    
     sync_parser = subparsers.add_parser('sync')
     sync_parser.add_argument('--config-file')
     sync_parser.add_argument('--config')
     sync_parser.add_argument('--dry', action='store_true', default=False)
     sync_parser.add_argument('--ntfy-topic', default=None, help='Tópico do ntfy para notificação')
     configure_parser = subparsers.add_parser('configure')
-
+    
     result = parser.parse_args()
-
+    
     if result.debug:
         logger.setLevel(logging.DEBUG)
-
+        logger.debug('Debug mode enabled')
+    
     if result.cmd == 'configure':
         init_config()
     elif result.cmd == 'sync':
         if result.config_file and result.config or not result.config_file and not result.config:
             raise Exception('É necessário informar um arquivo de configuração ou uma string de configuração')
-
+        
         if result.config_file:
             path = Path(result.config_file)
             if not path.exists():
                 raise Exception(f'Arquivo de configuração "{path}" não encontrado')
-
+            
             config = ImporterConfig.from_dict(json.loads(path.read_text()))
         else:
             config = ImporterConfig.from_dict(json.loads(base64.b64decode(result.config)))
         ntfy_topic = result.ntfy_topic
         sync(config, result.dry, ntfy_topic)
+    else:
+        parser.print_help()
 
 
 if __name__ == '__main__':

brbanks2ynab/importers/nubank/nubank_credit_card.py

@@ -12,11 +12,11 @@
 
 
 def _is_active_transactions(transaction: dict):
-    return transaction['details']['status'] == 'settled'
+    return transaction['details'].get('status', 'settled') == 'settled'
 
 
 def _is_installment_transaction(transaction: dict):
-    return 'charges' in transaction['details']
+    return 'charges' in transaction['details'] and transaction['details']['charges']['count'] > 1
 
 
 class NubankCreditCardData(DataImporter):
@@ -55,17 +55,20 @@ def _expand_installment_transaction(self, transaction: dict) -> List[Transaction
         parsed_date = datetime.strptime(transaction['time'][:10], '%Y-%m-%d')
 
         def _to_transaction(index) -> Transaction:
-            formatted_value = locale.currency(transaction['amount'] / 100, grouping=True)
+            formatted_value = locale.currency(transaction['amount'] / 100, grouping=True, symbol=False)
             # Adds the index to the date so the transactions spans multiple months
-            date = (parsed_date + relativedelta(months=index)).strftime('%Y-%m-%d')
+            date = (parsed_date + relativedelta(months=index - 1))
+            if index != 1:
+                # If it's not the first transaction, sets the day to the first
+                date = date.replace(day=1)
             return {
-                'transaction_id': f'{transaction["id"]}-{index}',
+                'transaction_id': f'{index}-{transaction["id"]}'[:35],
                 'account_id': self.account_id,
                 'payee': transaction['description'],
                 'amount': installment_amount,
-                'date': date,
-                'memo': f'Parcela {index} de {count}. Valor total: {formatted_value}',
-                'flag': 'Parcelado'
+                'date': date.strftime('%Y-%m-%d'),
+                'memo': f'Parcela {index} de {count}. Valor total: R$ {formatted_value}',
+                'flag': 'red'
             }
 
         return [_to_transaction(i + 1) for i in range(count)]

brbanks2ynab/sync/sync.py

@@ -9,8 +9,8 @@
 from brbanks2ynab.config.config import ImporterConfig
 from brbanks2ynab.importers import get_importers_for_bank
 from brbanks2ynab.util import find_budget_by_name
-from brbanks2ynab.ynab.ynab_transaction_importer import YNABTransactionImporter
 from brbanks2ynab.utils.notification import send_notification
+from brbanks2ynab.ynab.ynab_transaction_importer import YNABTransactionImporter
 
 logger = logging.getLogger('brbanks2ynab')
 logger.setLevel(logging.DEBUG)
@@ -24,37 +24,39 @@ def _build_summary(response: CreateTransactionResponse) -> dict:
     }
 
 
-def sync(config: ImporterConfig, dry: bool, notify: Optional[str] = None):
+def sync(config: ImporterConfig, dry: bool, ntfy_topic: Optional[str] = None):
     ynab = YNAB(config.ynab_token)
-
+    
     budget = find_budget_by_name(ynab.budgets.get_budgets().data.budgets, config.ynab_budget)
     ynab_accounts = ynab.accounts.get_accounts(budget.id).data.accounts
-
+    
     ynab_importer = YNABTransactionImporter(ynab, budget.id, config.start_import_date)
-
+    
     for bank in config.banks:
         importers = get_importers_for_bank(bank, config, ynab_accounts)
-
+        
         for importer in importers:
             ynab_importer.get_transactions_from(importer)
-
+    
     if dry:
         logger.info('Dry running! No transactions will be imported into YNAB.')
         logger.info(f'{len(ynab_importer.transactions)} would be imported into YNAB')
-
+        
         with open('import_result.json', 'w') as f:
             data = [dataclasses.asdict(t) for t in ynab_importer.transactions]
             json.dump(data, f)
     else:
         response = ynab_importer.save()
+        logger.debug(f'YNAB response: \n {json.dumps(dataclasses.asdict(response), indent=2)}')
+        
         summary = _build_summary(response)
-
+        
         logger.info(f"""
         {summary['transaction_count']} new transactions imported into YNAB
         {summary['duplicated_count']} transactions were already imported.
         """)
-        if notify:
-            send_notification(summary, notify)
+        if ntfy_topic:
+            send_notification(summary, ntfy_topic)
 
 
 if __name__ == '__main__':

brbanks2ynab/ynab/ynab_transaction_importer.py

@@ -14,26 +14,28 @@ def __init__(self, ynab: YNAB, budget_id: str, starting_date: str):
         self.budget_id = budget_id
         self.starting_date = datetime.strptime(starting_date, '%Y-%m-%d')
         self.transactions: List[TransactionRequest] = []
-
+    
     def get_transactions_from(self, transaction_importer: DataImporter):
         transactions = transaction_importer.get_data()
         transactions = filter(self._filter_transaction, transactions)
         transformed = map(self._create_transaction_request, transactions)
         self.transactions.extend(transformed)
         return self
-
+    
     def save(self):
         return self.ynab.transactions.create_transactions(self.budget_id, self.transactions)
-
+    
     def _create_transaction_request(self, transaction: Transaction) -> TransactionRequest:
         return TransactionRequest(
             transaction['account_id'],
             transaction['date'],
             transaction['amount'],
             payee_name=transaction['payee'],
             import_id=transaction['transaction_id'],
+            memo=transaction['memo'],
+            flag_color=transaction['flag'],
         )
-
+    
     def _filter_transaction(self, transaction: Transaction) -> bool:
         now = datetime.now()
         transaction_date = datetime.strptime(transaction['date'], '%Y-%m-%d')

requirements.txt

@@ -4,6 +4,7 @@ PyJWT==2.2.0
 
 ynab_sdk==0.5.0
 inquirer==2.8.0
+python-dateutil==2.8.2
 
 pytest==6.2.5
 pytest-cov==2.11.1

tests/importers/nubank/test_nubank_credit_card.py

@@ -60,9 +60,9 @@ def test_should_only_import_settled_transactions(setup_nubank, monkeypatch):
 def test_should_expand_installment_transactions(setup_nubank, monkeypatch):
     nu = setup_nubank
     installment_transaction = build_card_transaction({
+        'amount': 3000,
         'details': {
             'status': 'settled',
-            'amount': 3000,
             'charges': {
                 'count': 3,
                 'amount': 1000
@@ -83,11 +83,68 @@ def test_should_expand_installment_transactions(setup_nubank, monkeypatch):
 
     assert len(imported_transactions) == 4
     assert imported_transactions[0]['amount'] == -30000
+    assert imported_transactions[1]['memo'] == 'Parcela 1 de 3. Valor total: R$ 30,00'
+    assert imported_transactions[1]['flag'] == 'red'
     assert imported_transactions[1]['amount'] == -10000
+    assert imported_transactions[2]['memo'] == 'Parcela 2 de 3. Valor total: R$ 30,00'
+    assert imported_transactions[2]['flag'] == 'red'
     assert imported_transactions[2]['amount'] == -10000
+    assert imported_transactions[3]['memo'] == 'Parcela 3 de 3. Valor total: R$ 30,00'
+    assert imported_transactions[3]['flag'] == 'red'
     assert imported_transactions[3]['amount'] == -10000
     # All ids should be unique
     assert len(ids) == len(set(ids))
-    # Installments should have the memo and flag set
-    assert imported_transactions[1]['memo'] == 'Parcela 1 de 3. Valor total R$ 30,00'
-    assert imported_transactions[1]['flag'] == 'Red'
+    
+
+
+def test_should_not_expand_transactions_with_one_installment(setup_nubank, monkeypatch):
+    nu = setup_nubank
+    installment_transaction = build_card_transaction({
+        'amount': 3000,
+        'details': {
+            'status': 'settled',
+            'charges': {
+                'count': 1,
+                'amount': 3000
+            }
+        }
+    })
+    transactions = [
+        installment_transaction
+    ]
+    monkeypatch.setattr(nu, 'get_card_statements', lambda: transactions)
+    
+    importer = NubankCreditCardData(nu, 'some-id')
+    imported_transactions = list(importer.get_data())
+    
+    assert len(imported_transactions) == 1
+    assert imported_transactions[0]['amount'] == -30000
+    assert imported_transactions[0]['memo'] == ''
+    assert imported_transactions[0]['flag'] is None
+
+
+def test_should_set_the_day_to_first_for_following_installments(setup_nubank, monkeypatch):
+    nu = setup_nubank
+    installment_transaction = build_card_transaction({
+        'amount': 3000,
+        'time': '2021-01-15',
+        'details': {
+            'status': 'settled',
+            'charges': {
+                'count': 3,
+                'amount': 1000
+            }
+        }
+    })
+    transactions = [
+        installment_transaction
+    ]
+    monkeypatch.setattr(nu, 'get_card_statements', lambda: transactions)
+    
+    importer = NubankCreditCardData(nu, 'some-id')
+    imported_transactions = list(importer.get_data())
+    
+    assert len(imported_transactions) == 3
+    assert imported_transactions[0]['date'] == '2021-01-15'
+    assert imported_transactions[1]['date'] == '2021-02-01'
+    assert imported_transactions[2]['date'] == '2021-03-01'

tests/ynab/test_ynab_transaction_importer.py

@@ -13,7 +13,7 @@
 
 
 class FakeImporter(DataImporter):
-
+    
     def get_data(self) -> Iterable[Transaction]:
         return [
             {
@@ -22,36 +22,39 @@ def get_data(self) -> Iterable[Transaction]:
                 'payee': 'Some Payee',
                 'amount': 15000,
                 'date': today.strftime('%Y-%m-%d'),
+                'memo': '',
+                'flag': None,
             },
             {
                 'transaction_id': '2',
                 'account_id': 'some-id-two',
                 'payee': 'Some Mall',
                 'amount': 99000,
                 'date': (today - timedelta(weeks=54)).strftime('%Y-%m-%d'),
+                'memo': '',
+                'flag': None,
             }
         ]
 
 
 class TestYNABTransactionImporter(unittest.TestCase):
-
+    
     def test_should_import_transactions(self):
         fake_ynab = Mock(YNAB)
         start_date = (today - timedelta(weeks=4)).strftime('%Y-%m-%d')
         importer = YNABTransactionImporter(fake_ynab, '1234', start_date)
-
+        
         importer.get_transactions_from(FakeImporter())
         importer.save()
-
+        
         fake_ynab.transactions.create_transactions.assert_called_once()
-
+    
     def test_should_ignore_transactions_past_start_date(self):
         fake_ynab = Mock(YNAB)
         start_date = (today - timedelta(weeks=4)).strftime('%Y-%m-%d')
         importer = YNABTransactionImporter(fake_ynab, '1234', start_date)
-
+        
         importer.get_transactions_from(FakeImporter())
-
+        
         self.assertEqual(len(importer.transactions), 1)
         self.assertEqual(importer.transactions[0].import_id, '1')
-