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