# coding=utf-8 """Convert Bank statement CSV to Gnucash-acceptable format""" def valid_formats(): return _parsers.keys() def convert_file(fpath, fmt, encoding='utf-8'): try: parser_class = _parsers[fmt] except KeyError: raise ValueError('unsupported format: %s' % fmt) else: parser = parser_class() parser.parse_file(fpath, encoding) return parser.format_csv() def _unquote(value): if len(value) < 2: return value if value[0] == value[-1] == "'": return value[1:-1] if value[0] == value[-1] == '"': return value[1:-1] return value class BaseTransaction(object): """A CSV-imported transaction""" pass class GCTransaction(object): """A Gnucash-acceptable transaction""" def __init__(self, date, deposit, withdrawal, description): self.date = date self.deposit = deposit self.withdrawal = withdrawal self.description = description def __str__(self): """Naive CSV line printer""" values = [self.date, self.deposit, self.withdrawal, self.description] return ";".join(['"%s"' % v for v in values]) class BaseTransactionParser(object): """A statement parser""" name = "__NONE__" def __init__(self): self.transactions = [] self.state = False self.next_state = False def is_open(self): return self.state def is_closed(self): return not self.state def parse_file(self, fpath, encoding='utf-8'): lines_read = 0 with open(fpath) as f: for line in f.readlines(): line = line.decode(encoding).strip() lines_read += 1 self.state = self.pick_state(line, lineno=lines_read) if self.is_open(): self.transactions.append(self.parse_line(line).to_gc()) def format_csv(self): return "\n".join([ unicode(t).encode('utf-8') for t in self.transactions ]) class MbankTransaction(BaseTransaction): """Item in an Mbank CSV""" # Datum uskutečnění transakce # Datum zaúčtování transakce # Popis transakce # Zpráva pro příjemce # Plátce/Příjemce # Číslo účtu plátce/příjemce # KS # VS # SS # Částka transakce # Účetní zůstatek po transakci def __init__(self, line): fields = [self._cleanup_field(f) for f in line.split(";")] self.date_r = self._convert_date(fields.pop(0)) self.date_b = self._convert_date(fields.pop(0)) __ = fields.pop() self.new_balance = self._parse_currency(fields.pop()) amount = self._parse_currency(fields.pop()) if amount >= 0: self.amountd = amount self.amountw = 0 else: self.amountd = 0 self.amountw = -amount self._scrap = fields def _cleanup_field(self, field): x = ' '.join(_unquote(field).strip().split()) return '' if x == '-' else x def _convert_date(self, text): day, month, year = text.split('-') return '-'.join([year, month, day]) def _description(self): type_, message, party, number, ks, vs, ss = self._scrap message = message.replace(u' DATUM PROVEDENÍ TRANSAKCE: ', '; DATUM: ') out = [] if type_: out.append(type_) if message: out.append(message) if party: out.append(party) return " / ".join(out) def _parse_currency(self, text): """Read Mbank currency format""" num = text.replace(' ', '') num = num.replace(',', '.') return float(num) def to_gc(self): """Convert to GCTransaction""" return GCTransaction( date=self.date_r, deposit=self.amountd, withdrawal=self.amountw, description=self._description() ) class MbankTransactionParser(BaseTransactionParser): def parse_line(self, line): return MbankTransaction(line) def pick_state(self, line, lineno=None): """Choose parser state according to current line and line no.""" if self.next_state: self.next_state = False return True if self.is_closed() and line.startswith(u'#Datum uskute'): self.next_state = True return False if self.is_open() and not line: return False return self.state class FioTransaction(BaseTransaction): # ID operace # Datum # Objem # Měna # Protiúčet # Název protiúčtu # Kód banky # Název banky # KS # VS # SS # Poznámka # Zpráva pro příjemce # Typ # Provedl # Upřesnění # Poznámka # BIC # ID pokynu def __init__(self, line): fields = [self._cleanup_field(f) for f in line.split(";")] __ = fields.pop(0) self.date = self._convert_date(fields.pop(0)) amount = self._parse_currency(fields.pop(0)) if amount >= 0: self.amountd = amount self.amountw = 0 else: self.amountd = 0 self.amountw = -amount self.currency = fields.pop(0) assert self.currency == 'CZK' self._scrap = [self._cleanup_field(f) for f in fields] def _cleanup_field(self, field): x = ' '.join(_unquote(field).strip().split()) return '' if x == '-' else x def _description(self): ( party_acc, party_accnname, party_bankcode, party_bankname, ks, vs, ss, note1, message, type_, who, explanation, note2, bic, comm_id ) = self._scrap out = [] if message: out.append(message) return " / ".join(out) def _convert_date(self, text): day, month, year = text.split('.') return '-'.join([year, month, day]) def _parse_currency(self, text): return float(text) def to_gc(self): """Convert to GCTransaction""" return GCTransaction( date=self.date, deposit=self.amountd, withdrawal=self.amountw, description=self._description() ) class FioTransactionParser(BaseTransactionParser): def parse_line(self, line): return FioTransaction(line) def pick_state(self, line, lineno=None): """Choose parser state according to current line and line no.""" if self.next_state: self.next_state = False return True if self.is_closed() and line.startswith(u'"ID operace";"Datum";'): self.next_state = True return False if self.is_open() and line.startswith(u'"bankId";"2010"'): return False return self.state _parsers = { 'CzMbank': MbankTransactionParser, 'CzFio': FioTransactionParser, }