drebedengi.py

from datetime import datetime
from typing import List, Dict, NamedTuple, Iterator, Optional
from pathlib import Path

from kython.kompress import open as kopen

class Expense(NamedTuple):
    dt: datetime
    amount: float
    currency: str
    from_: str
    to: str # TODO to is basically category?
    comment: str
    eid: str
    amount_from: float
    currency_from: str


def _get_last_data() -> List[str]:
    BDIR = Path('/path/to/exports')
    lastz = max(BDIR.glob('drebedengi_*.zip'))
    fname = lastz.stem[len('drebedengi_'):] + '.txt'
    with kopen(lastz, fname) as fo:
        return [l.decode('utf-8').strip() for l in fo.readlines()]


def iter_expenses() -> Iterator[Expense]:
    lines = _get_last_data()

    d: Dict[str, List[str]] = {}
    currec = None
    for l in lines:
        if l.startswith('['):
            currec = l
        else:
            assert currec is not None
            if currec not in d:
                d[currec] = []
            ll = d[currec]
            ll.append(l)

    currencies = {
        curid: curname.strip('"')
        for curid, _, exchr, curname, _, _, _ in (
            l.split(';') for l in d['[currency]']
        )
    }
    INITIAL = '<initial>'
    objects: Dict[str, str] = {
        '-1': INITIAL,
    }
    for rec in d['[objects]']:
        spl = rec.split(';')
        oid = spl[0]
        desc = spl[3].strip('"')
        objects[oid] = desc

    rid = 0
    records = list(reversed(d['[records]']))
    while rid < len(records):
        amnts, currid, category, froms, dates, comment, _, empty = records[rid].split(';')
        amnt = int(amnts) / 100
        dtime = datetime.strptime(dates, "%Y-%m-%d %H:%M:%S")
        cmnt = comment.strip('"')
        currency = currencies[currid]
        eid = f'{dtime}-{amnt}-{rid}' # meh..
        from_ = objects[froms]
        to_ = objects[category]

        if rid + 1 < len(records):
            # try to handle exchange/move first
            namnts, ncurrid, ncategory, nfroms, _, _, _, _ = records[rid + 1].split(';')
            if ncategory == froms and nfroms == category:
                namnt = int(namnts) / 100
                ncurrency = currencies[ncurrid]

                if namnt < 0:
                    namnt, amnt = amnt, namnt
                    ncurrency, currency = currency, ncurrency # TODO erm?
                    from_, to_ = to_, from_

                # TODO map tags to something nicer??
                yield Expense(
                    dt=dtime,
                    amount=namnt,
                    currency=ncurrency,
                    to=to_,
                    amount_from=amnt,
                    currency_from=currency,
                    from_=from_,
                    comment=comment,
                    eid=eid,
                )
                rid += 2
                continue
        # otherwise normal transaction

        if amnt > 0:
            # drebedengi is weird...
            xx = to_
            to_ = from_
            from_ = xx
        else:
            amnt = -amnt
        assert amnt > 0
        yield Expense(
            dt=dtime,
            amount=amnt,
            currency=currency,
            to=to_,
            amount_from=-amnt,
            currency_from=currency,
            from_=from_,
            comment=cmnt,
            eid=eid,
        )
        rid += 1


def get_expenses() -> List[Expense]:
    return list(sorted(iter_expenses(), key=lambda e: e.dt))


def main():
    for e in get_expenses():
        print(e)


if __name__ == '__main__':
    main()