matrix_fridge_bot/db.py

   1 import logging
   2 import os
   3 from datetime import datetime as dt
   4 from datetime import timedelta
   5
   6 from sqlmodel import Field, Session, SQLModel, create_engine, select, text
   7
   8 ## Todo List
   9 # - [x] add to fridge
  10 # - [ ] reminder to eat recent food
  11 # - [ ] reminder to toss old food
  12
  13 DATE_FORMAT = '%A, %Y-%m-%d'
  14 MODULE='fridge'
  15 STALE_DAYS = 4
  16 TOSS_DAYS = 5
  17
  18 def today():
  19     return dt.now().strftime(DATE_FORMAT)
  20
  21 logger = logging.getLogger(__name__)
  22 logging.basicConfig(filename=f'matrix-{MODULE}-bot.log', encoding='utf-8', level=logging.DEBUG)
  23 logging.getLogger('nio').setLevel(logging.WARNING)
  24 logging.getLogger('asyncio').setLevel(logging.WARNING)
  25 logging.getLogger('invoke').setLevel(logging.WARNING)
  26
  27 class Food(SQLModel, table=True):
  28     id: int | None = Field(default=None, primary_key=True)
  29     name: str = Field(default=None, unique=True)
  30     added_date: dt = Field(default_factory=lambda: dt.now())
  31
  32     def __str__(self):
  33         added = self.added_date.strftime(DATE_FORMAT)
  34         return f'{self.name} ({added})'
  35
  36 class Data():
  37     def __init__(self, db_file):
  38
  39         self.file = db_file
  40         logger.debug(f'DB file: {db_file}')
  41         self.engine = self.get_engine()
  42
  43     def get_db_file(self):
  44         return self.file
  45
  46     def get_engine(self):
  47         return create_engine(f"sqlite:///{self.file}")
  48
  49     def setup(self):
  50         SQLModel.metadata.create_all(self.engine)
  51
  52     @staticmethod
  53     def get_stale():
  54         stale = dt.now() - timedelta(STALE_DAYS)
  55         logger.debug(f'Stale date is {stale}')
  56         return stale
  57
  58     def list(self):
  59         results = []
  60         with Session(self.engine) as session:
  61             stale_dt = Data.get_stale()
  62             cmd = select(Food).where(Food.added_date > stale_dt)
  63             results = session.exec(cmd).all()
  64             logger.debug(f'List restuls are {str(results)}')
  65             if(not results):
  66                 results.append('Nothing Found')
  67
  68         return results
  69
  70     def to_toss(self):
  71         results = []
  72         with Session(self.engine) as session:
  73             stale_dt = Data.get_stale()
  74             cmd = select(Food).where(Food.added_date <= stale_dt)
  75             results = session.exec(cmd).all()
  76             if(not results):
  77                 results.append('Nothing Found')
  78
  79         return results
  80
  81
  82     def filter(self, term):
  83         results = []
  84         for item in self.list():
  85             if term in item:
  86                 results.append(item)
  87
  88         return results
  89
  90     def add(self, name):
  91         try:
  92             with Session(self.engine) as session:
  93                 new = Food(name=name)
  94                 session.add(new)
  95                 session.commit()
  96             return f'Added {name}.'
  97         except:
  98             return f'Unable to add {name}. May be duplicate.'
  99
 100     @staticmethod
 101     def get_food(session, food):
 102         cmd = select(Food).filter(Food.name.ilike(f'%{food}%'))
 103         return session.exec(cmd).one()
 104
 105     def remove(self, name):
 106         # TODO: Tag as tossed or eaten
 107         with Session(self.engine) as session:
 108             item  = Data.get_food(session, name)
 109             if(not item):
 110                 return f'Not found: {name}'
 111
 112             try:
 113                 name = item.name
 114                 session.remove(item)
 115                 session.commit()
 116                 return f'Removed item {name}'
 117             except Exception as ex:
 118                 return f'Unable to remove item mark chore as done: {ex}'
 119
 120