Mercurial > louis > ofxstatement-us-hsbc
diff src/ofxstatement/plugins/us_hsbc/plugin.py @ 7:829eb62755b0
First cut at an HSBC (USA) plugin
author | Louis Opter <kalessin@kalessin.fr> |
---|---|
date | Thu, 17 Nov 2016 16:25:12 -0800 |
parents | |
children | 164da24a2997 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/ofxstatement/plugins/us_hsbc/plugin.py Thu Nov 17 16:25:12 2016 -0800 @@ -0,0 +1,156 @@ +# Copyright (c) 2016, Louis Opter <louis@opter.org> +# +# This file is part of ofxstatement-us-hsbc. +# +# ofxstatement-us-hsbc is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# ofxstatement-us-hsbc is distributed in the hope that it will be +# useful, but WITHOUT ANY WARRANTY; without even the implied warranty +# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +import contextlib +import csv +import datetime +import enum +import locale +import logging +import pdb +import sys + +from decimal import Decimal, Decimal as D +from ofxstatement.parser import CsvStatementParser +from ofxstatement.plugin import Plugin +from ofxstatement.statement import StatementLine, generate_transaction_id +from typing import Any, Dict, Iterable, List +from typing.io import TextIO + +from .record import CsvIndexes, Record +from .transactions import enrich + + +logger = logging.getLogger("ofxstatement.plugins.us_hsbc") + + +@contextlib.contextmanager +def _override_locale(category, value): + save = locale.getlocale(category) + locale.setlocale(category, value) + yield + locale.setlocale(category, save) + + +class Parser(CsvStatementParser): + + date_format = "%m/%d/%Y" + + mappings: Dict[str, int] = { + "date": CsvIndexes.DATE.value, + "amount": CsvIndexes.AMOUNT.value, + } + + _DESCRIPTION_PREFIX_TO_TRNTYPE: Dict[str, str] = { + "PURCHASE": "POS", + "RETURN": "POS", + "CASH WITHDRAWAL": "CASH", + "CHECK": "CHECK", + "CREDIT RECEIVED ON": "CREDIT", + "MISCELLANEOUS CREDIT": "CREDIT", + "CASH CONCENTRATION VENMO": "XFER", + "DEPOSIT FROM": "DIRECTDEP", + "DEPOSIT": "DEP", + "INTEREST EARNED AND PAID": "INT", + "ONLINE PAYMENT TO": "PAYMENT", + "PAY TO": "PAYMENT", + "PAYMENT -": "PAYMENT", # receiving a payment (e.g: on your CC account) + "PAYMENT TO": "PAYMENT", + "REBATE OF ATM SURCHARGE": "DEP", + "SERVICE CHG": "SRVCHG", + "ATM OR OTHER ELECTRONIC BANKING TRANSACTION": "ATM", + } + + def __init__(self, fin: TextIO) -> None: + super(Parser, self).__init__(fin) + + self.statement.currency = "USD" + self.statement.start_date = datetime.datetime.now() # timezones pls? + self.statement.end_date = self.statement.start_date + + def split_records(self) -> Iterable[str]: + return csv.reader(self.fin, delimiter=",", quotechar='"', strict=True) + + @classmethod + def _get_trntype(cls, record: Record) -> str: + for prefix, trntype in cls._DESCRIPTION_PREFIX_TO_TRNTYPE.items(): + if record.original_description.upper().startswith(prefix): + return trntype + + if record.simple_description.upper() == "TRANSFER": + return "XFER" + + logger.info( + "The transaction type for the following couldn't be determined and " + "will default to OTHER: {}".format(record) + ) + + return "OTHER" + + def parse_record(self, row: List[str]) -> StatementLine: + if self.cur_record < 4: # starts at 1 + logger.info("Skipping row: {}".format(row)) + return None # skip (all) the csv headers + + try: + sl = super(Parser, self).parse_record(row) + record = Record(row) + + if record.currency != "$": + logger.warning( + "Skipping record {} ({}) which doesn't appear to be in " + "USD".format(self.cur_record, ",".join(row)) + ) + return None + + sl.trntype = self._get_trntype(record) + enrich(sl, record) + sl.id = generate_transaction_id(sl) + + self.statement.start_date = min(sl.date, self.statement.start_date) + self.statement.end_date = max(sl.date, self.statement.end_date) + except Exception: + logger.exception("Parsing failed:") + logger.info("Press {} to exit the debugger".format( + "^Z" if sys.platform.startswith("win32") else "^D" + )) + pdb.post_mortem() + sys.exit(1) + + return sl + + @_override_locale(locale.LC_NUMERIC, "en_US") + def parse_decimal(self, value: str) -> Decimal: + return D(locale.delocalize(value)) + + +class HSBCUSAPlugin(Plugin): + + DEFAULT_CHARSET = "latin1" + + def get_parser(self, filename: str) -> Parser: + encoding = self.settings.get("charset", HSBCUSAPlugin.DEFAULT_CHARSET) + + # XXX: how does this gets closed? + fin = open(filename, "r", encoding=encoding) + + parser = Parser(fin) + parser.statement.bank_id = self.settings.get("routing_number") + parser.statement.account_id = self.settings.get("account_number") + parser.statement.account_type = self.settings.get("account_type", "CHECKING") + + return parser