diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..1b8d5f9 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/converter.py b/src/converter.py similarity index 74% rename from converter.py rename to src/converter.py index 332f010..0b33d98 100644 --- a/converter.py +++ b/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'] diff --git a/src/tests/__init__.py b/src/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/tests/test_converter.py b/src/tests/test_converter.py new file mode 100644 index 0000000..7c959ee --- /dev/null +++ b/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' + +