aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rwxr-xr-xscripts/ofx2json115
1 files changed, 115 insertions, 0 deletions
diff --git a/scripts/ofx2json b/scripts/ofx2json
new file mode 100755
index 0000000..08d9fe9
--- /dev/null
+++ b/scripts/ofx2json
@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+# Copyright 2019 vg@devys.org
+# SPDX-License-Identifier: MIT
+
+progname = 'ofx2json'
+
+# be explicit about __doc__ to allow progname to be defined first
+__doc__ = f'''
+Convert ofx from input file (or stdin if no FILENAME) to json on stdout.
+
+Usage: {progname} [options] [--] [FILENAME...]
+ {progname} -h|--help
+
+Options:
+ -h, --help Display this help message
+ -n option doing n
+ -N ARG option doing N with ARG
+'''
+
+
+### standard modules
+import collections
+import datetime
+import decimal
+import io
+import json
+import os
+import subprocess
+import sys
+### external modules
+import docopt
+import ofxparse
+
+# This script is not compatible below python3.7.2, always abort (not in main
+# since the syntax itself can cause the script to fail later inconveniently).
+assert sys.hexversion >= 0x03070200
+
+
+def map_ofx_transaction(transaction):
+ return collections.OrderedDict(
+ amount=transaction.amount,
+ checknum=transaction.checknum,
+ date=transaction.date,
+ id=transaction.id,
+ mcc=transaction.mcc,
+ memo=transaction.memo,
+ payee=transaction.payee,
+ sic=transaction.sic,
+ type=transaction.type,
+ )
+
+
+def map_ofx_statement(statement):
+ return collections.OrderedDict(
+ available_balance=statement.available_balance,
+ available_balance_date=statement.available_balance_date,
+ balance=statement.balance,
+ balance_date=statement.balance_date,
+ currency=statement.currency,
+ discarded_entries=statement.discarded_entries,
+ end_date=statement.end_date,
+ start_date=statement.start_date,
+ transactions=[map_ofx_transaction(i)
+ for i in statement.transactions],
+ warnings=statement.warnings,
+ )
+
+
+def map_ofx_account(account):
+ return collections.OrderedDict(
+ account_id=account.account_id,
+ account_type=account.account_type,
+ branch_id=account.branch_id,
+ curdef=account.curdef,
+ institution=account.institution,
+ number=account.number,
+ rounting_number=account.routing_number,
+ statement=map_ofx_statement(account.statement),
+ type=account.type,
+ warnings=account.warnings,
+ )
+
+
+def map_ofx(ofx):
+ return collections.OrderedDict(
+ headers=ofx.headers,
+ accounts=[map_ofx_account(i) for i in ofx.accounts],
+ )
+
+
+def parse_ofx(filepaths):
+ if not filepaths:
+ yield ofxparse.OfxParser.parse(io.StringIO(sys.stdin.read()))
+ else:
+ for filepath in filepaths:
+ with open(filepath) as fh:
+ yield ofxparse.OfxParser.parse(fh)
+
+
+def main():
+ 'function called only when script invoked directly on command line'
+ args = docopt.docopt(__doc__)
+
+ def json_default(obj):
+ if isinstance(obj, decimal.Decimal):
+ return str(obj) # no rounding by dumping a string, not float
+ elif hasattr(obj, 'isoformat'):
+ return obj.isoformat()
+ print(f'error for {obj}')
+ raise TypeError
+
+ print(json.dumps(list(map(map_ofx, parse_ofx(args['FILENAME']))),
+ indent=2, default=json_default))
+
+main()