Added checks

Author: ccauet

ledgertools/__init__.py

@@ -0,0 +1,28 @@
+import os
+import math
+
+class CheckFailed(Exception):
+    '''raise this when there's at least one check does not pass'''
+
+def valid_accounts():
+    try:
+        with open('.valid_accounts', 'r') as infile:
+            lines = infile.readlines()
+        return [line.rstrip() for line in lines]
+    except:
+        print('No ".valid_accounts" file found!')
+        return []
+
+def transaction_checks():
+    return [check_sum_is_zero, check_valid_accounts]
+
+def check_sum_is_zero(transaction, abs_tol=0.001):
+    sum_of_postings = sum([p.amount for p in transaction.postings])
+    return math.isclose(sum_of_postings, 0, abs_tol=abs_tol), f'Sum of postings is off by {sum_of_postings:.2f}'
+
+def check_valid_accounts(transaction):
+    invalid_accounts = []
+    for posting in transaction.postings:
+        if posting.account not in valid_accounts():
+            invalid_accounts.append(posting.account)
+    return len(invalid_accounts) < 1, f'Invalid account names: {invalid_accounts}'

ledgertools/checks.py

@@ -15,10 +15,11 @@ def cli():
 @click.option('-f', '--file', 'in_file', help='Input file name', prompt='Input file name')
 @click.option('-n', '--name', default='transactions.json', help='Output file name')
 @click.option('-p', '--pickle', 'as_pickle', is_flag=True, help='Output as pickle file')
+@click.option('--run-checks', 'run_checks', is_flag=True, help='Run standard checks on data')
 @click.option('--stdout', is_flag=True, help='Output to stdout, supresses output files')
-def read(in_file, name, as_pickle, stdout):
+def read(in_file, name, as_pickle, run_checks, stdout):
     click.secho(f'Reading input file: {in_file}', fg='green')
-    transactions = read_file(in_file)
+    transactions = read_file(in_file, run_checks)
 
     if stdout:
         print(json.dumps(transactions, sort_keys=True, indent=4, ensure_ascii=False))

ledgertools/cli.py

@@ -4,6 +4,8 @@
 import pendulum
 
 from tqdm import tqdm
+from ledgertools.checks import transaction_checks, CheckFailed
+
 
 # regex copied from https://github.com/Tagirijus/ledger-parse
 PAT_TRANSACTION_DATA = re.compile(r'(?P<year>\d{4})[/|-](?P<month>\d{2})[/|-](?P<day>\d{2})(?:=(?P<year_aux>\d{4})[/|-](?P<month_aux>\d{2})[/|-](?P<day_aux>\d{2}))? (?P<state>[\*|!])?[ ]?(\((?P<code>[^\)].+)\) )?(?P<payee>.+)')
@@ -15,10 +17,10 @@ class Ledger():
     """
     def __init__(self, ledger_filename=None, raw_transactions=None):
         self._transactions = []
-        if raw_transactions is not None:
-            self._transactions = self._import_raw_transactions(raw_transactions)
         if ledger_filename is not None:
             self._transactions = self.import_ledger_file(ledger_filename)
+        if raw_transactions is not None:
+            self._transactions = self._import_raw_transactions(raw_transactions)
 
     @property
     def json(self):
@@ -28,6 +30,17 @@ def json(self):
     def transactions(self):
         return self._transactions
 
+    def run_checks(self, strict=True):
+        n_checks_failed = 0
+        for transaction in self._transactions:
+            for check in transaction_checks():
+                success, info = transaction.run_checks(check)
+                if not success:
+                    n_checks_failed += 1
+                    print(f'{check.__name__} failed for transaction\n {transaction}{info}')
+        if n_checks_failed > 0 and strict:
+            raise CheckFailed(f'{n_checks_failed} checks failed')
+
     def _parse_tags(self, string):
         tags = {}
         # split string in parts
@@ -187,7 +200,7 @@ def json(self):
                     status=self.status,
                     code=self.code,
                     description=self.description,
-                    transactions=[p.json for p in self.postings],
+                    postings=[p.json for p in self.postings],
                     comment=self.comment)
 
     @property
@@ -234,6 +247,9 @@ def postings(self):
     def add_posting(self, posting):
         self._postings.append(posting)
 
+    def run_checks(self, check):
+        return check(self)
+
 
 class Posting():
     """Represents a single posting
@@ -275,5 +291,8 @@ def date(self):
             return self._primary_date.isoformat()
 
 
-def read_file(in_file):
-    return Ledger(ledger_filename=in_file).json
+def read_file(in_file, run_checks=False):
+    ledger = Ledger(ledger_filename=in_file)
+    if run_checks:
+        ledger.run_checks(strict=False)
+    return ledger.json