grocery-manager/grocery_transactions.py

#!/usr/bin/python3
#
# Copyright (c) Daniel Sheffield 2021
#
# All rights reserved
#
# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
import time
import itertools
import sys
from typing import Union
import urwid
from urwid import raw_display
from app.db_utils import (
    QueryManager,
    get_insert_product_statement
)
from app.activities import (
    show_or_exit,
    ActivityManager,
)  
from app.activities.TransactionEditor import TransactionEditor
from app.activities.NewProduct import NewProduct
from app.widgets import AutoCompleteEdit, AutoCompleteFloatEdit

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

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

dark_palette = [
    ("popup_focus", "light red", "light gray"),
    ('popup', 'light red', 'dark gray'),
    ('banner', 'light gray', 'dark red'),
    ('streak', 'light red', 'dark gray'),
    ('bg', 'light red', 'black'),
]

light_palette = [
    ("popup_focus", "dark blue", "light gray"),
    ('popup', 'light blue', 'dark gray'),
    ('banner', 'light gray', 'dark red'),
    ('streak', 'white', 'dark gray'),
    ('bg', 'white', 'dark blue'),
]

grid_layout = [
    [ 'ts',          'store',    ],
    [ 'organic',     'product',  ],
    [ 'category',    'group',   ],
]
side_pane = [
    'unit',
    'quantity',
    'price',
]
bottom_pane = [
    'description',
    'dbview',
]

cols = [
    c for c in filter(
        lambda x: x is not None,
        itertools.chain(
            *grid_layout, side_pane, set(bottom_pane) - set(['dbview'])
        )
    )
]

display_map = {
    'ts': lambda x: f"{time.strftime('%Y-%m-%d %H:%M', (x.year, x.month, x.day, x.hour, x.minute, 0, 0, 0, 0))}",
    'price': lambda x: f"{x:.2f}",
    'quantity': lambda x: f"{x:.2f}",
    'organic': lambda x: "true" if x else "false",
}
display = lambda data, name: display_map[name](data) if name in display_map else data

def _apply_choice_callback(
    activity_manager: ActivityManager,
    base: str, name: str, value: str
):
    base = activity_manager.get(base)
    base.apply_choice(name, value)
    activity_manager.show(base.update())

def _insert_new_product_callback(activity_manager, query_manager, product, category, group):
    activity_manager.app.log.write(
        '{};\n'.format(get_insert_product_statement(product, category, group)))
    query_manager.insert_new_product(product, category, group)
    activity_manager.show(activity_manager.get('transaction').update())

def _new_product_callback(
    activity_manager: ActivityManager,
    query_manager: QueryManager,
    name: str, data: dict
):
    cur = activity_manager.current()
    txn : TransactionEditor = activity_manager.get('transaction')
    activity_manager.show(
        activity_manager.create(NewProduct, 'new_product',
            query_manager, cur, name, txn.data,
            lambda w, t, data: _autocomplete_callback(activity_manager, query_manager, w, t, data),
            txn.apply_changes,
            lambda product, category, group: _insert_new_product_callback(
                activity_manager, query_manager, product, category, group),
            lambda: activity_manager.show(cur.update()),
            lambda name, value: _apply_choice_callback(activity_manager, 'new_product', name, value)
        )
    )

def _autocomplete_callback(
    activity_manager: ActivityManager,
    query_manager: QueryManager,
    widget: Union[AutoCompleteEdit, AutoCompleteFloatEdit],
    name: str, data: dict
):
    options = query_manager.unique_suggestions(name, **data)
    if len(options) > 0:
        widget._emit('open', options)
    elif len(options) == 0 and activity_manager.current() is not activity_manager.get('new_product'):
        if name in ('product', 'category', 'group'):
            _new_product_callback(activity_manager, query_manager, name, data)


def _save_and_clear_callback(activity_manager):
    txn = activity_manager.get('transaction')
    activity_manager.app.save(txn.data)
    txn.clear()
    txn.focus_on_product()
    activity_manager.show(txn.update())

args = sys.argv
log = args[1]

class GroceryTransactionEditor(urwid.WidgetPlaceholder):
    def __init__(self, activity_manager, cur, log):
        super().__init__(urwid.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"
                #print(self.cur.mogrify(line))
                #input()
                self.cur.execute(line)
        
            if None not in (date, store):
                txn.apply_choice('ts', date)
                txn.apply_choice('store', store)

        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 = data['organic'] if data['organic'] else 'false'
        statement = \
            f"CALL insert_transaction('{ts}', $store${store}$store$, " \
            f"$descr${description}$descr$, {quantity}, $unit${unit}$unit$, " \
            f"{price}, $produ${product}$produ$, {organic});\n"
        self.log.write(statement)
        self.cur.execute(statement)

cur.execute("BEGIN")

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

activity_manager.create(TransactionEditor, 'transaction',
    query_manager, cols, grid_layout, side_pane, bottom_pane,
    lambda: _save_and_clear_callback(activity_manager),
    lambda widget, name, data: _autocomplete_callback(activity_manager, query_manager, widget, name, data),
    lambda name, value: _apply_choice_callback(activity_manager, 'transaction', name, value),
)

app = None
def iter_palettes():
    palettes = [dark_palette, light_palette]
    while True:
        p = palettes.pop(0)
        palettes.append(p)
        yield p

palettes = iter_palettes()
palette = next(palettes)

try:
    app = GroceryTransactionEditor(activity_manager, cur, log)
    screen = raw_display.Screen()
    loop = urwid.MainLoop(app, palette, 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()