Mercurial > louis > ofxstatement-us-hsbc

comparison src/ofxstatement/plugins/us_hsbc/plugin.py @ 9:28548158a325 default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Some minor improvements/fixes This will eventually evolve to something more generic, since I have re-used the same design, and some of the parts here, to write a plugin for Charles Schwab.
author Louis Opter <louis@opter.org>
date Thu, 09 Mar 2017 22:55:02 -0800
parents 164da24a2997
children
comparison
equal deleted inserted replaced
8:164da24a2997 9:28548158a325
16 # along with this program. If not, see <http://www.gnu.org/licenses/>. 16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17 17
18 import contextlib 18 import contextlib
19 import csv 19 import csv
20 import datetime 20 import datetime
21 import enum
22 import locale 21 import locale
23 import logging 22 import logging
24 import pdb 23 import pdb
25 import sys 24 import sys
26 25
27 from decimal import Decimal, Decimal as D 26 from decimal import Decimal, Decimal as D
28 from ofxstatement.parser import CsvStatementParser 27 from ofxstatement.parser import CsvStatementParser
29 from ofxstatement.plugin import Plugin 28 from ofxstatement.plugin import Plugin
30 from ofxstatement.statement import StatementLine, generate_transaction_id 29 from ofxstatement.statement import StatementLine, generate_transaction_id
31 from typing import Any, Dict, Iterable, List 30 from typing import Dict, Generator, Iterable, List
32 from typing.io import TextIO 31 from typing.io import TextIO
33 32
34 from .record import CsvIndexes, Record 33 from .record import CsvIndexes, Record
35 from .transactions import enrich 34 from .transactions import enrich
36 35
37 36
38 logger = logging.getLogger("ofxstatement.plugins.us_hsbc") 37 logger = logging.getLogger("ofxstatement.plugins.us_hsbc")
39 38
40 39
41 @contextlib.contextmanager 40 @contextlib.contextmanager
42 def _override_locale(category, value): 41 def _override_locale(category: int, value: str) -> Generator[None, None, None]:
43 save = locale.getlocale(category) 42 save = locale.getlocale(category)
44 locale.setlocale(category, value) 43 locale.setlocale(category, value)
45 yield 44 yield
46 locale.setlocale(category, save) 45 locale.setlocale(category, save)
46
47
48 class _spawn_debugger_on_exception:
49
50 def __init__(self, errmsg: str) -> None:
51 self._errmsg = errmsg
52
53 def __enter__(self) -> None:
54 pass
55
56 def __exit__(self, exc_type, exc_value, exc_tb) -> bool:
57 if exc_value is not None:
58 logger.exception(self._errmsg)
59 logger.info("Press {} to exit the debugger".format(
60 "^Z" if sys.platform.startswith("win32") else "^D"
61 ))
62 pdb.post_mortem()
63 sys.exit(1)
47 64
48 65
49 class Parser(CsvStatementParser): 66 class Parser(CsvStatementParser):
50 67
51 date_format = "%m/%d/%Y" 68 date_format = "%m/%d/%Y"
63 "CREDIT RECEIVED ON": "CREDIT", 80 "CREDIT RECEIVED ON": "CREDIT",
64 "MISCELLANEOUS CREDIT": "CREDIT", 81 "MISCELLANEOUS CREDIT": "CREDIT",
65 "CASH CONCENTRATION VENMO": "XFER", 82 "CASH CONCENTRATION VENMO": "XFER",
66 "DEPOSIT FROM": "DIRECTDEP", 83 "DEPOSIT FROM": "DIRECTDEP",
67 "DEPOSIT": "DEP", 84 "DEPOSIT": "DEP",
85 "HSBC SECURITIES": "XFER",
68 "INTEREST EARNED AND PAID": "INT", 86 "INTEREST EARNED AND PAID": "INT",
69 "ONLINE PAYMENT TO": "PAYMENT", 87 "ONLINE PAYMENT TO": "PAYMENT",
70 "PAY TO": "PAYMENT", 88 "PAY TO": "PAYMENT",
71 "PAYMENT -": "PAYMENT", # receiving a payment (e.g: on your CC account) 89 "PAYMENT -": "PAYMENT", # receiving a payment (e.g: on your CC account)
72 "PAYMENT TO": "PAYMENT", 90 "PAYMENT TO": "PAYMENT",
104 def parse_record(self, row: List[str]) -> StatementLine: 122 def parse_record(self, row: List[str]) -> StatementLine:
105 if self.cur_record < 4: # starts at 1 123 if self.cur_record < 4: # starts at 1
106 logger.info("Skipping row: {}".format(row)) 124 logger.info("Skipping row: {}".format(row))
107 return None # skip (all) the csv headers 125 return None # skip (all) the csv headers
108 126
109 try: 127 with _spawn_debugger_on_exception("Parsing failed:"):
110 sl = super(Parser, self).parse_record(row) 128 sl = super(Parser, self).parse_record(row)
111 record = Record(row) 129 record = Record(row)
112 130
113 if record.currency != "$": 131 if record.currency != "$":
114 logger.warning( 132 logger.warning(
121 enrich(sl, record) 139 enrich(sl, record)
122 sl.id = generate_transaction_id(sl) 140 sl.id = generate_transaction_id(sl)
123 141
124 self.statement.start_date = min(sl.date, self.statement.start_date) 142 self.statement.start_date = min(sl.date, self.statement.start_date)
125 self.statement.end_date = max(sl.date, self.statement.end_date) 143 self.statement.end_date = max(sl.date, self.statement.end_date)
126 except Exception:
127 logger.exception("Parsing failed:")
128 logger.info("Press {} to exit the debugger".format(
129 "^Z" if sys.platform.startswith("win32") else "^D"
130 ))
131 pdb.post_mortem()
132 sys.exit(1)
133 144
134 return sl 145 return sl
135 146
136 @_override_locale(locale.LC_NUMERIC, "en_US") 147 @_override_locale(locale.LC_NUMERIC, "en_US")
137 def parse_decimal(self, value: str) -> Decimal: 148 def parse_decimal(self, value: str) -> Decimal:
146 encoding = self.settings.get("charset", HSBCUSAPlugin.DEFAULT_CHARSET) 157 encoding = self.settings.get("charset", HSBCUSAPlugin.DEFAULT_CHARSET)
147 158
148 # XXX: how does this gets closed? 159 # XXX: how does this gets closed?
149 fin = open(filename, "r", encoding=encoding) 160 fin = open(filename, "r", encoding=encoding)
150 161
151 parser = Parser(fin) 162 p = Parser(fin)
152 parser.statement.bank_id = self.settings.get("routing_number") 163 p.statement.bank_id = self.settings.get("routing_number")
153 parser.statement.account_id = self.settings.get("account_number") 164 p.statement.account_id = self.settings.get("account_number")
154 parser.statement.account_type = self.settings.get("account_type", "CHECKING") 165 p.statement.account_type = self.settings.get("account_type", "CHECKING")
155 166
156 return parser 167 return p