matrix_shop_bot/db.py

   1 import logging
   2 import os
   3 from datetime import datetime as dt
   4 from decimal import Decimal
   5 from typing import List
   6
   7 from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
   8
   9 from sqlite3 import IntegrityError
  10
  11 MODULE='shop'
  12
  13 DATE_FORMAT = '%Y-%m-%d'
  14
  15 def today():
  16     return dt.now().strftime(DATE_FORMAT)
  17
  18
  19 logger = logging.getLogger(__name__)
  20
  21 class ItemStoreLink(SQLModel, table=True):
  22     item_id: int | None = Field(default=None, foreign_key="item.id", primary_key=True)
  23     store_id: int | None = Field(default=None, foreign_key="store.id", primary_key=True)
  24
  25
  26 class Item(SQLModel, table=True):
  27     id: int | None = Field(default=None, primary_key=True)
  28     name: str = Field(default=None, unique=True)
  29     price: Decimal | None = Field(decimal_places=2)
  30     stores: list["Store"] = Relationship(back_populates="shopping_list", link_model=ItemStoreLink)
  31     purchase_date: dt | None = Field(default=None)
  32     def __str__(self):
  33         return f'{self.name} (#{self.id})'
  34
  35 class PreferStore(SQLModel, table=True):
  36     store_id: int | None = Field(default=None, foreign_key="store.id", primary_key=True)
  37
  38 class Store(SQLModel, table=True):
  39     id: int | None = Field(default=None, primary_key=True)
  40     name: str = Field(default=None, unique=True)
  41     shopping_list: list["Item"] = Relationship(back_populates="stores", link_model=ItemStoreLink)
  42     # purchase_history: list["PurchaseRecord"] = Relationship(back_populates="store")
  43     def __str__(self):
  44         return f'{self.name} (#{self.id})'
  45
  46 # May get used later...
  47 class PurchaseRecord(SQLModel, table=True):
  48     id: int | None = Field(default=None, primary_key=True)
  49     name: str = Field(default=None)
  50     price: Decimal | None = Field(decimal_places=2)
  51     store: str | None = Field(default=None)
  52     purchase_date: dt | None = Field(default=None)
  53
  54 def _get_with_id(cls, item_id):
  55     return select(cls).where(cls.id == item_id)
  56
  57
  58 class Data():
  59     def __init__(self, db_file=None):
  60         if not db_file:
  61             db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), f'{MODULE}.db')
  62
  63         self.file = db_file
  64         logger.debug(f'Using db file {self.file}')
  65         self.engine = self.get_engine()
  66         self.last_store_id = None
  67
  68     def _set_pref_store_id(self, store_id):
  69         with Session(self.engine) as session:
  70             prefs = session.exec(select(PreferStore)).all()
  71             for pref in prefs:
  72                 session.delete(pref)
  73             pref = PreferStore()
  74             pref.store_id = store_id
  75             session.add(pref)
  76             session.commit()
  77
  78     def _get_pref_store_id(self):
  79         with Session(self.engine) as session:
  80             pref = session.exec(select(PreferStore)).one()
  81             return pref.store_id
  82
  83     def _add_with_name(self, cls, name):
  84         item_id = -1
  85         try:
  86             with Session(self.engine) as session:
  87                 new = cls(name=name)
  88                 session.add(new)
  89                 session.commit()
  90                 item_id = new.id
  91         except:
  92             # duplicate add
  93             # recover by returning the item_id
  94             item = session.exec(select(Item).where(Item.name == name)).first()
  95             item_id = item.id
  96
  97         return item_id
  98
  99     def add_store(self, name):
 100         store_id = self._add_with_name(Store, name)
 101         self._set_pref_store_id(store_id)
 102         return store_id
 103
 104     def add_item(self, name):
 105         names = name.split(',')
 106         for name in names:
 107             item_id = self._add_with_name(Item, name.strip())
 108             store_id = self._get_pref_store_id()
 109             self.assign_store(item_id, store_id)
 110         return item_id
 111
 112     def assign_last_store(self, item_id):
 113         store_id = self._get_pref_store_id()
 114         self.assign_store(item_id, store_id)
 115         return store_id
 116
 117     def assign_store(self, item_id, store_id):
 118         self._set_pref_store_id(store_id)
 119
 120         with Session(self.engine) as session:
 121             store = session.exec(_get_with_id(Store, store_id)).first()
 122             item = session.exec(_get_with_id(Item, item_id)).first()
 123             try:
 124                 item.stores.append(store)
 125                 session.commit()
 126             except:
 127                 pass # Already assigned
 128         return ['Success']
 129
 130     def list_purchases(self, store_name):
 131         with Session(self.engine) as session:
 132             cmd = select(PurchaseRecord).where(PurchaseRecord.store == store_name)
 133             records = session.exec(cmd).all()
 134             return records
 135
 136     def purchase_items(self, item_ids):
 137         results = []
 138         with Session(self.engine) as session:
 139             # Mark items as purchased
 140             for item_id in item_ids.split(' '):
 141                 item = session.exec(_get_with_id(Item, item_id)).first()
 142                 if item:
 143                     item.purchase_date = dt.today()
 144                 results.append(str(item))
 145
 146                 # Track purchase history (locally only!)
 147                 # to generate recommended shopping lists
 148                 record = PurchaseRecord()
 149                 record.name = item.name
 150                 record.price = item.price
 151                 record.purchase_date = item.purchase_date
 152                 record.store = self._get_pref_store_id() # We bought it here
 153                 session.add(record)
 154
 155             session.commit()
 156         return results
 157
 158
 159     def clear_list(self, store_id, clear_all=False):
 160         with Session(self.engine) as session:
 161             store = session.exec(_get_with_id(Store, store_id)).first()
 162             for item in store.shopping_list:
 163                 if clear_all or item.purchase_date:
 164                     session.delete(item)
 165             session.commit()
 166
 167     def get_stores(self):
 168         with Session(self.engine) as session:
 169             cmd = select(Store)
 170             return session.exec(cmd).all()
 171
 172     def get_unsorted_list(self):
 173
 174         results = []
 175         with Session(self.engine) as session:
 176             cmd = select(Item).where(Item.stores == None)
 177             items = session.exec(cmd).all()
 178
 179             if not items:
 180                 return [f'No unsorted items!']
 181             return items
 182
 183
 184     def get_shopping_list_for_store(self, store_id) -> List[str]:
 185         self._set_pref_store_id(store_id)
 186         return self.get_shopping_list(store_id)
 187
 188     def get_last_shopping_list(self) -> List[str]:
 189         store_id = self._get_pref_store_id()
 190         return self.get_shopping_list(store_id)
 191
 192     def get_shopping_list(self, store_id) -> List[str]:
 193         results = []
 194         with Session(self.engine) as session:
 195             cmd = select(Store).where(Store.id == store_id)
 196             store = session.exec(cmd).first()
 197             if not store:
 198                 return [f'Store {store_id} not found.']
 199
 200             results.append(f' ## {store.name}')
 201
 202             items = store.shopping_list
 203             if not items:
 204                 results.append(f'-- Nothing to shop for at {store.name} --')
 205             else:
 206                 for item in items:
 207                     if(item.purchase_date):
 208                         results.append(f' - [x] {item}')
 209                     else:
 210                         results.append(f' - [ ] {item}')
 211         return results
 212
 213     def remove_item(self, item_id):
 214         with Session(self.engine) as session:
 215             item = session.exec(_get_with_id(Item, item_id)).first()
 216             session.delete(item)
 217             session.commit()
 218
 219     def remove_store(self, store_id):
 220         with Session(self.engine) as session:
 221             store = session.exec(_get_with_id(Store, store_id)).first()
 222             session.delete(store)
 223             session.commit()
 224
 225     def get_engine(self):
 226         return create_engine(f"sqlite:///{self.file}")
 227
 228     def setup(self):
 229         SQLModel.metadata.create_all(self.engine)
 230         return 'Success'
 231
 232
 233
 234