feat: Add ntfy support

brbanks2ynab/cli.py

@@ -27,6 +27,7 @@ def main():
     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()
@@ -48,8 +49,8 @@ def main():
             config = ImporterConfig.from_dict(json.loads(path.read_text()))
         else:
             config = ImporterConfig.from_dict(json.loads(base64.b64decode(result.config)))
-
-        sync(config, result.dry)
+        ntfy_topic = result.ntfy_topic
+        sync(config, result.dry, ntfy_topic)
 
 
 if __name__ == '__main__':

brbanks2ynab/importers/itau/itau_checking_account.py

@@ -19,18 +19,17 @@ def get_data(self) -> Iterable[Transaction]:
         return map(self._to_transaction, transactions)
 
     def _to_transaction(self, itau_transaction: dict) -> Transaction:
-        # Parses the date in this format: 11/01/2023
         date = datetime.strptime(itau_transaction['dataLancamento'], '%d/%m/%Y')
         amount = float(itau_transaction['valorLancamento'].replace('.', '').replace(',', '.'))
         is_inflow = itau_transaction['ePositivo']
         description = itau_transaction['descricaoLancamento']
-        tx_id = hashlib.sha1(f'{date.isoformat()}:{amount}:{is_inflow}:{description}'.encode('utf-8')).hexdigest()
+        tx_id = str(hashlib.sha1(f'{date.isoformat()}:{amount}:{is_inflow}:{description}'.encode('utf-8')).hexdigest())
 
         if not is_inflow:
             amount *= -1
 
         return {
-            'transaction_id': str(tx_id),
+            'transaction_id': tx_id[:35],
             'account_id': self.account_id,
             'payee': description,
             'amount': int(amount * 1000),

brbanks2ynab/sync/sync.py

@@ -1,20 +1,30 @@
 import dataclasses
 import json
 import logging
-from pathlib import Path
+from typing import Optional
 
 from ynab_sdk import YNAB
+from ynab_sdk.api.models.responses.transactions import CreateTransactionResponse
 
 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
 
 logger = logging.getLogger('brbanks2ynab')
 logger.setLevel(logging.DEBUG)
 
 
-def sync(config: ImporterConfig, dry: bool):
+def _build_summary(response: CreateTransactionResponse) -> dict:
+    return {
+        'transaction_count': len(response.transaction_ids),
+        'duplicated_count': len(response.duplicate_import_ids),
+        'total_amount': sum(t.amount for t in response.transactions) / 1000,
+    }
+
+
+def sync(config: ImporterConfig, dry: bool, notify: Optional[str] = None):
     ynab = YNAB(config.ynab_token)
 
     budget = find_budget_by_name(ynab.budgets.get_budgets().data.budgets, config.ynab_budget)
@@ -35,10 +45,16 @@ def sync(config: ImporterConfig, dry: bool):
         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.info(f'{len(response["data"]["transaction_ids"])} transactions imported')
+        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 __name__ == '__main__':

brbanks2ynab/utils/notification.py

@@ -0,0 +1,18 @@
+import requests
+
+
+def send_notification(summary: dict, topic):
+    message = f"""
+{summary['transaction_count']} new transactions imported into YNAB
+
+{summary['duplicated_count']} transactions were already imported.
+    """
+
+    payload = {
+        'topic': topic,
+        'title': f'💰 R$ {summary["transaction_count"]} new transactions imported into YNAB',
+        'message': message,
+        'tags': ['cron', 'ynab'],
+    }
+
+    requests.post('https://ntfy.sh', json=payload)

brbanks2ynab/ynab/ynab_transaction_importer.py

@@ -35,6 +35,7 @@ def _create_transaction_request(self, transaction: Transaction) -> TransactionRe
         )
 
     def _filter_transaction(self, transaction: Transaction) -> bool:
+        now = datetime.now()
         transaction_date = datetime.strptime(transaction['date'], '%Y-%m-%d')
 
-        return transaction_date >= self.starting_date
+        return self.starting_date <= transaction_date <= now

requirements.txt

@@ -3,7 +3,7 @@ git+https://github.com/andreroggeri/pybradesco.git@80450d483dac1e6cde02bd9520ed9
 git+https://github.com/andreroggeri/python-alelo.git@a03cf8d483ffe40eeac4d00030e0d913e569cb70
 PyJWT==2.2.0
 
-ynab_sdk==0.2.4
+ynab_sdk==0.5.0
 inquirer==2.7.0
 
 pytest==6.2.3