gcconv/gcconv.py

# coding=utf-8
"""Convert Bank statement CSV to Gnucash-acceptable format"""

from collections import namedtuple

ParserState = namedtuple('ParserState', 'ths nxt')


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 = ParserState(ths=False, nxt=False)

    def is_open(self):
        return self.state.ths

    def is_closed(self):
        return not self.state.ths

    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.advance(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"""

    # (a) Datum uskutečnění transakce
    # (b) Datum zaúčtování transakce
    # (c) Popis transakce
    # (d) Zpráva pro příjemce
    # (e) Plátce/Příjemce
    # (f) Číslo účtu plátce/příjemce
    # (g) KS
    # (h) VS
    # (i) SS
    # (j) Částka transakce
    # (k) Účetní zůstatek po transakci
    # (l) (empty column at the end)

    def __init__(self, line):
        fields = [self._cleanup_field(f) for f in line.split(";")]
        self.date_r = self._convert_date(fields.pop(0))         # (a)
        self.date_b = self._convert_date(fields.pop(0))         # (b)
        __ = fields.pop()                                       # (l)
        self.new_balance = self._parse_currency(fields.pop())   # (k)
        amount = self._parse_currency(fields.pop())             # (j)
        if amount >= 0:
            self.amountd = amount
            self.amountw = 0
        else:
            self.amountd = 0
            self.amountw = -amount
        self._scrap = fields                # (c-i)

    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
        dpt = u' DATUM PROVEDENÍ TRANSAKCE: '
        if dpt in message:
            message, __ = message.split(dpt)
        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 advance(self, line, lineno=None):
        """Choose parser state according to current line and line no."""
        if self.state.nxt:
            return ParserState(ths=True, nxt=False)
        if self.is_closed() and line.startswith(u'#Datum uskute'):
            return ParserState(ths=False, nxt=True)
        if self.is_open() and not line:
            return ParserState(ths=False, nxt=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 advance(self, line, lineno=None):
        """Choose parser state according to current line and line no."""
        if self.state.nxt:
            return ParserState(ths=True, nxt=False)
        if self.is_open() and line.startswith(u'"ID operace";"Datum";'):
            return ParserState(ths=False, nxt=True)
        return self.state


_parsers = {
    'CzMbank': MbankTransactionParser,
    'CzFio': FioTransactionParser,
}