Mercurial > louis > ofxstatement-us-hsbc
view src/ofxstatement/plugins/us_hsbc/plugin.py @ 9:28548158a325 default tip
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 |
line wrap: on
line source
# 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 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 Dict, Generator, 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: int, value: str) -> Generator[None, None, None]: save = locale.getlocale(category) locale.setlocale(category, value) yield locale.setlocale(category, save) class _spawn_debugger_on_exception: def __init__(self, errmsg: str) -> None: self._errmsg = errmsg def __enter__(self) -> None: pass def __exit__(self, exc_type, exc_value, exc_tb) -> bool: if exc_value is not None: logger.exception(self._errmsg) logger.info("Press {} to exit the debugger".format( "^Z" if sys.platform.startswith("win32") else "^D" )) pdb.post_mortem() sys.exit(1) 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", "HSBC SECURITIES": "XFER", "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 record 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 with _spawn_debugger_on_exception("Parsing failed:"): 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) 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) p = Parser(fin) p.statement.bank_id = self.settings.get("routing_number") p.statement.account_id = self.settings.get("account_number") p.statement.account_type = self.settings.get("account_type", "CHECKING") return p