import logging
import os
from datetime import datetime as dt
+from datetime import timedelta
from sqlmodel import Field, Session, SQLModel, create_engine, select, text
## Todo List
-# - [ ] add Vision FK
-# - [ ] add notes to each chore
+# - [ ] add to fridge
+# - [ ] reminder to eat recent food
+# - [ ] reminder to toss old food
-DATE_FORMAT = '%Y-%m-%d'
+DATE_FORMAT = '%A, %Y-%m-%d'
+MODULE='fridge'
+STALE_DAYS = 2
+TOSS_DAYS = 4
def today():
return dt.now().strftime(DATE_FORMAT)
logger = logging.getLogger(__name__)
-logging.basicConfig(filename='chore.log', encoding='utf-8', level=logging.DEBUG)
+logging.basicConfig(filename=f'matrix-{MODULE}-bot.log', encoding='utf-8', level=logging.DEBUG)
-class Chore(SQLModel, table=True):
+class Food(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
+ added_date: dt = Field(default_factory=lambda: dt.now())
def __str__(self):
- return f'{self.name} ({self.id})'
-
-
-class Record(SQLModel, table=True):
- id: int | None = Field(default=None, primary_key=True)
- chore_id: int | None = Field(default=None, foreign_key="chore.id")
- complete_date: dt = Field(default_factory=lambda: dt.now())
-
-
-def get_chore(session, chore_id):
- cmd = select(Chore).where(Chore.id == chore_id )
- return session.exec(cmd).first()
+ added = self.added_date.strftime(DATE_FORMAT)
+ return f'{self.name} ({added})'
class Data():
def __init__(self, db_file):
def setup(self):
SQLModel.metadata.create_all(self.engine)
-
@staticmethod
- def check_chore(session, chore_id) -> str:
- chore = get_chore(session, chore_id)
- if(not chore):
- return f'Chore not found: {chore_id}'
-
- sql = text('''
- SELECT * FROM Record
- WHERE chore_id = :chore_id
- AND complete_date >= :today
- ORDER BY complete_date DESC
- LIMIT 1
- ''')
- params = {}
- params['chore_id'] = chore_id
- params['today'] = today()
- results = session.execute(sql, params).first()
-
- logger.debug(results)
- if(results):
- return f'- [x] {chore}'
- else:
- return f'- [ ] {chore}'
-
-
- def check(self, chore_id):
- with Session(self.engine) as session:
- return Data.check_chore(session, chore_id)
+ def get_stale():
+ stale = dt.now() - timedelta(STALE_DAYS)
+ logger.debug(f'Stale date is {stale}')
+ return stale
def list(self):
results = []
with Session(self.engine) as session:
- cmd = select(Chore)
- chores = session.exec(cmd).all()
- if(not chores):
- results.append('None Found')
+ stale_dt = Data.get_stale()
+ cmd = select(Food).where(Food.added_date > stale_dt)
+ results = session.exec(cmd).all()
+ logger.debug(f'List restuls are {str(results)}')
+ if(not results):
+ results.append('Nothing Found')
- for chore in chores:
- results.append(Data.check_chore(session, chore.id))
return results
+ def to_toss(self):
+ results = []
+ with Session(self.engine) as session:
+ stale_dt = Data.get_stale()
+ cmd = select(Food).where(Food.added_date <= stale_dt)
+ results = session.exec(cmd).all()
+ if(not results):
+ results.append('Nothing Found')
+
+ return results
+
+
def filter(self, term):
results = []
for item in self.list():
return results
-
-
-
def add(self, name):
try:
with Session(self.engine) as session:
- new_chore = Chore(name=name)
- session.add(new_chore)
+ new = Food(name=name)
+ session.add(new)
session.commit()
- return f'Created chore {name}.'
+ return f'Added {name}.'
except:
- return f'Unable to create {name}. May be duplicate.'
-
+ return f'Unable to add {name}. May be duplicate.'
+ @staticmethod
+ def get_food(session, food):
+ cmd = select(Food).filter(Food.name.ilike(f'%{food}%'))
+ return session.exec(cmd).one()
- def done(self, chore_id):
+ def remove(self, name):
+ # TODO: Tag as tossed or eaten
with Session(self.engine) as session:
- chore = get_chore(session, chore_id)
- if(not chore):
- return f'Chore not found: {chore_id}'
+ item = Data.get_food(session, name)
+ if(not item):
+ return f'Not found: {name}'
try:
- session.add(Record(chore_id=chore.id))
+ name = item.name
+ session.remove(item)
session.commit()
- return f'Marked chore as done: {chore.name}.'
+ return f'Removed item {name}'
except Exception as ex:
- return f'Unable to mark chore as done: {ex}'
+ return f'Unable to remove item mark chore as done: {ex}'
- def undo(self, chore_id):
- with Session(self.engine) as session:
- cmd = select(Chore).where(Chore.id == chore_id)
- chore = session.exec(cmd).first()
-
- if(not chore):
- return f'Not found: {chore_id}'
-
- try:
- cmd = select(Record).where(Record.chore_id == chore_id).where(Record.complete_date >= today())
- results = session.exec(cmd).all()
- for result in results:
- session.delete(result)
- session.commit()
- return f'Marked chore as *not* done: {chore.name}.'
- except Exception as ex:
- return f'Unable to mark chore as done: {ex}'
-
- def remove(self, chore_id):
- try:
- with Session(self.engine) as session:
- chore = get_chore(session, chore_id)
- if(not chore):
- return f'Chore not found: {chore_id}'
-
- sql = text('''
- DELETE FROM Record
- WHERE chore_id = :chore_id
- ''')
- params = {}
- params['chore_id'] = chore_id
- results = session.execute(sql, params)
- logger.debug(results)
- sql = text('''
- DELETE FROM Chore
- WHERE id = :chore_id
- ''')
- results = session.execute(sql, params)
- logger.debug(results)
- session.commit()
- return f'Removed chore {chore_id}'
- except Exception as ex:
- return f'Unable to remove chore: {ex}'
projects / matrix-fridge-bot.git / blobdiff