Initial commit for moneymanager output

pytest.ini

@@ -0,0 +1,12 @@
+[pytest]
+env =
+    D:DB_HOST=''
+    D:DB_PORT=1
+    D:DB_NAME=''
+
+mongodb_fixture_dir =
+    src/tests/mongo_fixtures
+
+mongodb_fixtures =
+    imported_transactions
+    accounts

src/converter.py

@@ -16,7 +16,7 @@ import time
 import watchdog.events
 import watchdog.observers
 
-DATE_FORMAT = "%d/%m/%Y"
+MB_DATE_FORMAT = "%d/%m/%Y"
 
 if 'WATCH_DIR' in os.environ:
     WATCH_DIR = os.environ['WATCH_DIR']
@@ -34,20 +34,41 @@ MONGO_COL = 'imported_transactions'
 ACCOUNT_COL = 'accounts'
 
 MONGO_URL = "mongodb://{}:{}".format(MONGO_URL, MONGO_PORT)
-myclient = pymongo.MongoClient(MONGO_URL)
-mydb = myclient[MONGO_DB]
-mongo_col = mydb[MONGO_COL]
-account_col = mydb[ACCOUNT_COL]
 
 logging.basicConfig(format='%(asctime)s - %(message)s', level=logging.INFO)
 logging.basicConfig(format='ERROR: %(asctime)s - %(message)s', level=logging.ERROR)
 
 class Handler(watchdog.events.PatternMatchingEventHandler):
-    def __init__(self):
+    mydb = None
+
+    def __init__(self, mongo_db=None):
         # Set the patterns for PatternMatchingEventHandler
         watchdog.events.PatternMatchingEventHandler.__init__(self, patterns=['*.qfx'],
                                                              ignore_directories=True, case_sensitive=False)
 
+        if mongo_db is None:
+            myclient = pymongo.MongoClient(MONGO_URL)
+            mydb = myclient[MONGO_DB]
+        else:
+            self.mydb = mongo_db
+
+        #mongo_col = mydb[MONGO_COL]
+        #account_col = mydb[ACCOUNT_COL]
+  
+    @staticmethod
+    def line_to_moneybrilliant(line):
+        return {
+                'date': line['date'].strftime(MB_DATE_FORMAT),
+                'memo' : line['memo'],
+                'category': 'Uncategorised',
+                'amount': line['amount'],
+                'name': line['name']
+                }
+
+    @staticmethod
+    def line_to_mmex(line):
+        return ''
+
     @staticmethod
     def write_csv(statement, out_file):
         logging.info("Writing: " + out_file)
@@ -62,11 +83,27 @@ class Handler(watchdog.events.PatternMatchingEventHandler):
             f.write("\r\n")
             writer = DictWriter(f, fieldnames=fields)
             for line in statement:
-                writer.writerow(line)
+                writer.writerow(Handler.line_to_moneybrilliant(line))
+
 
     @staticmethod
-    def transaction_exists(line):
-        existing_trans = mongo_col.find_one(line)
+    def write_mmex(statement, outfile):
+        logging.info("Writing: " + out_file)
+
+        if len(statement) == 0:
+            logging.info("No transactions to write.")
+            return
+
+        fields = ['date', 'payee', 'amount', 'category', 'subcategory', 'number', 'notes']
+        with open(out_file, 'w') as f:
+            f.write("Date,Payee,Amount,Category,Sub Category,Number,Notes")
+            f.write("\r\n")
+            writer = DictWriter(f, fieldnames=fields)
+            for line in statement:
+                writer.writerow(line)
+
+    def transaction_exists(self, line):
+        existing_trans = self.mydb[MONGO_COL].find_one(line)
 
         return existing_trans is not None
 
@@ -87,10 +124,9 @@ class Handler(watchdog.events.PatternMatchingEventHandler):
 
         return dict_item
 
-    @staticmethod
-    def get_statement_from_qfx(qfx):
+    def get_statement_from_qfx(self, qfx):
 
-        account = account_col.find_one({"number": qfx.account.number})
+        account = self.mydb[ACCOUNT_COL].find_one({"number": qfx.account.number})
         if account is None:
             logging.error("No account for account number {} exists. Create one and re-process the file".format(qfx.account.number))
 
@@ -101,21 +137,25 @@ class Handler(watchdog.events.PatternMatchingEventHandler):
                 continue
 
             line = {
-                'date': transaction.date.strftime(DATE_FORMAT),
+                'id': transaction.id,
+                'date': transaction.date,#.strftime(DATE_FORMAT),
                 'memo' : transaction.memo,
-                'category': 'Uncategorised',
+                #'category': 'Uncategorised',
                 'amount': transaction.amount,
-                'name': account['name']
+                'name': account['name'],
+                'payee': transaction.payee,
+                'type': transaction.type
             }
 
-            #mongo needs the decimal values in Decimal128, so create a version for it
-            line_d128 = Handler.convert_decimal(line.copy())
 
-            if Handler.transaction_exists(line_d128):
+            #mongo needs the decimal values in Decimal128, so create a version for it
+            line_d128 = self.convert_decimal(line.copy())
+
+            if self.transaction_exists(line_d128):
                 continue
 
             statement.append(line)
-            result = mongo_col.insert_one(line_d128)
+            result = self.mydb[MONGO_COL].insert_one(line_d128)
             logging.info("New db entry stored: {}".format(result.inserted_id))
 
         return statement, account['name']    

src/tests/test_converter.py

@@ -0,0 +1,45 @@
+from  ..converter import Handler
+
+from pathlib import Path
+from ofxparse import OfxParser
+from datetime import datetime
+from decimal import Decimal, getcontext
+
+import os
+
+class TestConverter:
+    def test_get_statement_from_qfx(self, mongodb):
+        
+        test_file = Path(os.getcwd()) / 'src' / 'tests' / 'sampleTrans1.qfx'
+
+        with open(test_file, 'r') as file:
+            qfx = OfxParser.parse(file, fail_fast=False)
+
+            handler = Handler(mongodb)
+
+            statement, account = handler.get_statement_from_qfx(qfx)
+
+            assert account == 'HSBC Everyday Global'
+            assert len(statement) == mongodb['imported_transactions'].find({}).count()
+
+
+    def test_line_to_moneybrilliant(self):
+        samples = [
+                    {'id': '202201900001', 'date': '20220119', 'memo': 'GOLDEN PRAWN 15JAN22... GENERATED', 'amount': 1.04, 'name': 'HSBC Everyday Global', 'payee': '2% CASHBACK - ENJOY', 'type': 'credit'},
+                    {'id': '202201900003', 'date': '20220119', 'memo': '19JAN22 ATMA896 08:2...843015 ATM', 'amount': -52.20, 'name': 'HSBC Everyday Global', 'payee': 'GOLDEN PRAWN', 'type': 'debit'},
+                    {'id': '202201400001', 'date': '20220114', 'memo': '14JAN22 127196  18:3...892226 ATM', 'amount': -23.15, 'name': 'HSBC Everyday Global', 'payee': 'ALDI STORES - DARRA', 'type': 'debit'}
+                    ]
+        
+        sample = samples[0]
+        sample['date'] = datetime.strptime(sample['date'], "%Y%m%d")
+        sample['amount'] = round(Decimal(sample['amount']), 2)
+        result = Handler.line_to_moneybrilliant(sample)
+
+        assert len(result) == 5
+        assert result['date'] == '19/01/2022'
+        assert result['memo'] == 'GOLDEN PRAWN 15JAN22... GENERATED'
+        assert result['category'] == 'Uncategorised'
+        assert result['amount'] == Decimal('1.04')
+        assert result['name'] == 'HSBC Everyday Global'
+
+