grocery-manager/grocery_transactions.py

#!/usr/bin/python3
#
# Copyright (c) Daniel Sheffield 2021 - 2023
#
# All rights reserved
#
# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
import sys
from psycopg import Cursor
from urwid import raw_display, WidgetPlaceholder, SolidFill, MainLoop
from app.activities.ActivityManager import ActivityManager, show_or_exit
from app.activities.TransactionEditor import TransactionEditor
from app.data.QueryManager import QueryManager, display_mapper
from app.palette import iter_palettes, solarized

try:
    from db_credentials import HOST, PASSWORD
    host = f'host={HOST}'
    password = f'password={PASSWORD}'
except:
    host = ''
    password = ''

try:
    import os

    import psycopg
    user = os.getenv('USER')
    conn: Cursor = psycopg.connect(f"{host} dbname=grocery user={user} {password}")
    cur = conn.cursor()
except:
    print('Failed to set up db connection. Entering Mock mode')
    from mock import *

args = sys.argv
log = args[1]

class GroceryTransactionEditor(WidgetPlaceholder):
    def __init__(self, activity_manager, cur, log):
        super().__init__(SolidFill(u'/'))
        self.activity_manager = activity_manager
        self.cur = cur
        txn: TransactionEditor = self.activity_manager.get('transaction')

        with open(log, 'r') as f:
            date = None
            store = None
            for line in f.readlines():
                if date is None and store is None:
                    if '$store$' in line:
                        date, store, _= line.split('$store$')
                        date = date.split("'")[1]
                else:
                    assert None not in (date, store,), \
                        "Both date and store should be set or neither should be set"
                    if '$store$' in line:
                        assert date in line and f'$store${store}$store$' in line, \
                            "Date ({date}) and store ({store}) not found in {line}."\
                            " Mixing transactions from different dates and stores is not supported"
                self.cur.execute(line)

            if None not in (date, store):
                txn.apply_choice('ts', date)
                txn.apply_choice('store', store)
            else:
                txn.focus_on(txn.edit_fields['ts'])

        self.activity_manager.show(self)
        self.activity_manager.show(txn.update())

        self.log = self.open(log)

    def _open(self, log):
        with open(log, 'a') as f:
            yield f

    def open(self, log):
        self._to_close = self._open(log)
        return next(self._to_close)

    def close(self):
        if self._to_close is not None:
            self._to_close.close()
            self._to_close = None

    def save(self, data):
        ts = data['ts']
        store = data['store']
        description = data['description']
        quantity = data['quantity']
        unit = data['unit']
        price = data['price']
        product = data['product']
        organic = 'true' if data['organic'] is True else 'false'
        tags = ', '.join([
            f'$quot${v}$quot$' for t,v in data.items() if 'tags' in t and v
        ])
        tags = f", ARRAY[{tags}]" if tags else ''
        statement = \
            f"CALL insert_transaction('{ts}', $store${store}$store$, " \
            f"$descr${description}$descr$, {quantity}, $unit${unit}$unit$, " \
            f"{price}, $produ${product}$produ$, {organic}{tags});\n"
        self.log.write(statement)
        self.log.flush()
        self.cur.execute(statement)

cur.execute("BEGIN")

activity_manager = ActivityManager()
query_manager = QueryManager(cur, display_mapper)

activity_manager.create(TransactionEditor, 'transaction',
    activity_manager, query_manager,
)

app = None

palettes = iter_palettes(solarized.theme)

try:
    app = GroceryTransactionEditor(activity_manager, cur, log)
    screen = raw_display.Screen()
    loop = MainLoop(app, next(palettes), screen=screen,
        unhandled_input=lambda k: show_or_exit(k, screen=screen, palettes=palettes),
        pop_ups=True
    )

    loop.run()

finally:
    if app is not None:
        app.close()

cur.close()
conn.close()