view src/ofxstatement/plugins/us_hsbc/plugin.py @ 8:164da24a2997

Minor adjustments Made those changes a couple weeks ago, can't exactly remember why :s.
author Louis Opter <kalessin@kalessin.fr>
date Thu, 01 Dec 2016 17:14:11 -0800
parents 829eb62755b0
children 28548158a325
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 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 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

        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