Mercurial > louis > ofxstatement-us-hsbc

comparison src/ofxstatement/plugins/us_hsbc/plugin.py @ 7:829eb62755b0

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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
comparison
equal deleted inserted replaced
6:5692e2a61764 7:829eb62755b0
1 # Copyright (c) 2016, Louis Opter <louis@opter.org>
2 #
3 # This file is part of ofxstatement-us-hsbc.
4 #
5 # ofxstatement-us-hsbc is free software: you can redistribute it and/or
6 # modify it under the terms of the GNU General Public License as
7 # published by the Free Software Foundation, either version 3 of the
8 # License, or (at your option) any later version.
9 #
10 # ofxstatement-us-hsbc is distributed in the hope that it will be
11 # useful, but WITHOUT ANY WARRANTY; without even the implied warranty
12 # of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18 import contextlib
19 import csv
20 import datetime
21 import enum
22 import locale
23 import logging
24 import pdb
25 import sys
26
27 from decimal import Decimal, Decimal as D
28 from ofxstatement.parser import CsvStatementParser
29 from ofxstatement.plugin import Plugin
30 from ofxstatement.statement import StatementLine, generate_transaction_id
31 from typing import Any, Dict, Iterable, List
32 from typing.io import TextIO
33
34 from .record import CsvIndexes, Record
35 from .transactions import enrich
36
37
38 logger = logging.getLogger("ofxstatement.plugins.us_hsbc")
39
40
41 @contextlib.contextmanager
42 def _override_locale(category, value):
43 save = locale.getlocale(category)
44 locale.setlocale(category, value)
45 yield
46 locale.setlocale(category, save)
47
48
49 class Parser(CsvStatementParser):
50
51 date_format = "%m/%d/%Y"
52
53 mappings: Dict[str, int] = {
54 "date": CsvIndexes.DATE.value,
55 "amount": CsvIndexes.AMOUNT.value,
56 }
57
58 _DESCRIPTION_PREFIX_TO_TRNTYPE: Dict[str, str] = {
59 "PURCHASE": "POS",
60 "RETURN": "POS",
61 "CASH WITHDRAWAL": "CASH",
62 "CHECK": "CHECK",
63 "CREDIT RECEIVED ON": "CREDIT",
64 "MISCELLANEOUS CREDIT": "CREDIT",
65 "CASH CONCENTRATION VENMO": "XFER",
66 "DEPOSIT FROM": "DIRECTDEP",
67 "DEPOSIT": "DEP",
68 "INTEREST EARNED AND PAID": "INT",
69 "ONLINE PAYMENT TO": "PAYMENT",
70 "PAY TO": "PAYMENT",
71 "PAYMENT -": "PAYMENT", # receiving a payment (e.g: on your CC account)
72 "PAYMENT TO": "PAYMENT",
73 "REBATE OF ATM SURCHARGE": "DEP",
74 "SERVICE CHG": "SRVCHG",
75 "ATM OR OTHER ELECTRONIC BANKING TRANSACTION": "ATM",
76 }
77
78 def __init__(self, fin: TextIO) -> None:
79 super(Parser, self).__init__(fin)
80
81 self.statement.currency = "USD"
82 self.statement.start_date = datetime.datetime.now() # timezones pls?
83 self.statement.end_date = self.statement.start_date
84
85 def split_records(self) -> Iterable[str]:
86 return csv.reader(self.fin, delimiter=",", quotechar='"', strict=True)
87
88 @classmethod
89 def _get_trntype(cls, record: Record) -> str:
90 for prefix, trntype in cls._DESCRIPTION_PREFIX_TO_TRNTYPE.items():
91 if record.original_description.upper().startswith(prefix):
92 return trntype
93
94 if record.simple_description.upper() == "TRANSFER":
95 return "XFER"
96
97 logger.info(
98 "The transaction type for the following couldn't be determined and "
99 "will default to OTHER: {}".format(record)
100 )
101
102 return "OTHER"
103
104 def parse_record(self, row: List[str]) -> StatementLine:
105 if self.cur_record < 4: # starts at 1
106 logger.info("Skipping row: {}".format(row))
107 return None # skip (all) the csv headers
108
109 try:
110 sl = super(Parser, self).parse_record(row)
111 record = Record(row)
112
113 if record.currency != "$":
114 logger.warning(
115 "Skipping record {} ({}) which doesn't appear to be in "
116 "USD".format(self.cur_record, ",".join(row))
117 )
118 return None
119
120 sl.trntype = self._get_trntype(record)
121 enrich(sl, record)
122 sl.id = generate_transaction_id(sl)
123
124 self.statement.start_date = min(sl.date, self.statement.start_date)
125 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
134 return sl
135
136 @_override_locale(locale.LC_NUMERIC, "en_US")
137 def parse_decimal(self, value: str) -> Decimal:
138 return D(locale.delocalize(value))
139
140
141 class HSBCUSAPlugin(Plugin):
142
143 DEFAULT_CHARSET = "latin1"
144
145 def get_parser(self, filename: str) -> Parser:
146 encoding = self.settings.get("charset", HSBCUSAPlugin.DEFAULT_CHARSET)
147
148 # XXX: how does this gets closed?
149 fin = open(filename, "r", encoding=encoding)
150
151 parser = Parser(fin)
152 parser.statement.bank_id = self.settings.get("routing_number")
153 parser.statement.account_id = self.settings.get("account_number")
154 parser.statement.account_type = self.settings.get("account_type", "CHECKING")
155
156 return parser