1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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()
|
