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, delete
   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 FORGET_DAYS = 10
  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     @staticmethod
  59     def get_forget():
  60         forget = dt.now() - timedelta(FORGET_DAYS)
  61         return forget
  62
  63
  64     def list(self):
  65         results = []
  66         with Session(self.engine) as session:
  67             stale_dt = Data.get_stale()
  68             cmd = select(Food).where(Food.added_date > stale_dt)
  69             results = session.exec(cmd).all()
  70             logger.debug(f'List restuls are {str(results)}')
  71             if(not results):
  72                 results.append('Nothing Found')
  73
  74         return results
  75
  76     def to_toss(self):
  77         results = []
  78         with Session(self.engine) as session:
  79             stale_dt = Data.get_stale()
  80             cmd = select(Food).where(Food.added_date <= stale_dt)
  81             results = session.exec(cmd).all()
  82             if(not results):
  83                 results.append('Nothing Found')
  84
  85         return results
  86
  87     def cleanup(self):
  88
  89         logger.debug(f'Cleaning up...')
  90         with Session(self.engine) as session:
  91             stale_dt = Data.get_forget()
  92             cmd = delete(Food).where(Food.added_date < stale_dt)
  93             result = session.exec(cmd)
  94             session.commit()
  95         return True
  96
  97
  98     def filter(self, term):
  99         results = []
 100         for item in self.list():
 101             if term in item:
 102                 results.append(item)
 103
 104         return results
 105
 106     def add(self, name):
 107         logger.debug(f'Attempting to add {name}')
 108         self.cleanup()  # Old items can block new ones
 109         try:
 110             with Session(self.engine) as session:
 111                 new = Food(name=name)
 112                 session.add(new)
 113                 session.commit()
 114             return f'Added {name}.'
 115         except:
 116             return f'Unable to add {name}. May be duplicate.'
 117
 118     @staticmethod
 119     def get_food(session, food):
 120         cmd = select(Food).filter(Food.name.ilike(f'%{food}%'))
 121         return session.exec(cmd).one()
 122
 123     def remove(self, name):
 124         # TODO: Tag as tossed or eaten
 125         with Session(self.engine) as session:
 126             item  = Data.get_food(session, name)
 127             if(not item):
 128                 return f'Not found: {name}'
 129
 130             try:
 131                 name = item.name
 132                 session.remove(item)
 133                 session.commit()
 134                 return f'Removed item {name}'
 135             except Exception as ex:
 136                 return f'Unable to remove item mark chore as done: {ex}'
 137
 138