--- /dev/null
+# Python does that
+__pycache__
+
+# Makefile
+.venv
+*.egg-info
+build
+
+# Debug logs
+*.log
+
+# Data file
+*.db
+
+# Chatbot output
+session.txt
--- /dev/null
+from datetime import datetime as dt
+import logging
+import os
+
+import simplematrixbotlib as botlib
+
+from .db import Data, pretty_print
+
+SERVER = os.environ.get('MATRIX_SERVER')
+BOT_NAME = os.environ.get('GOAL_BOT')
+BOT_PWD = os.environ.get('GOAL_BOT_PWD')
+
+def get_bot():
+ creds = botlib.Creds(SERVER, BOT_NAME, BOT_PWD)
+ return botlib.Bot(creds)
+
+DATE_FORMAT = '%Y-%m-%d'
+
+def today():
+ return dt.now().strftime(DATE_FORMAT)
+
+PREFIX = '!goal'
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(filename='goals.log', encoding='utf-8', level=logging.DEBUG)
+logging.getLogger('nio').setLevel(logging.WARNING)
+
+async def do_goal(match):
+ '''
+ Prefix: !goal
+
+ Synax:
+ !goal COMMAND ARGS
+
+ Commands:
+ help - show this help
+ add - add a goal
+ add-task ID - add a task under the goal specified by ID
+ add-vision - add a new vision (collection of goals)
+ close-goal ID - mark goal done by ID
+ close-vision ID - mark vision as complete by ID
+ done ID - mark task done by ID
+ list - list incomplete goals
+ list-all - list everything
+ list-done - list completed goals
+ print GOAL_ID - print a goal and all tasks
+ print-vision ID - print a vision and all goals
+ set-vision GOAL_ID VISION_ID- associate GOAL_ID with VISION_ID
+ visions - list incomplete visions
+ '''
+
+ if match.command('setup'):
+ result = Data().setup()
+ return [result]
+
+ if match.command('add'):
+ new = ' '.join(match.args()[1:])
+ result = Data().add_goal(new)
+ return [f'Goal added with id {result}']
+
+ if match.command('list'):
+ goals = Data().get_all_goals()
+ results = ['# Goals list...']
+ for goal in goals:
+ results.append(str(goal))
+ return results
+
+
+ if match.command('list-done'):
+ goals = Data().get_done_goals()
+ results = ['# Completed Goals...']
+ for goal in goals:
+ results.append(str(goal))
+ return results
+
+
+ if match.command('add-task'):
+ goal_id = int(match.args()[1])
+ new = ' '.join(match.args()[2:])
+ result = Data().add_task(goal_id=goal_id, name=new)
+ goal,tasks = Data().get_goal(goal_id)
+ results = pretty_print(goal, tasks)
+ return results
+ # return [f'Added task with id {result}']
+
+ if match.command('add-vision'):
+ new = ' '.join(match.args()[1:])
+ result = Data().add_vision(new)
+ return [f'Vision added with id {result}']
+
+ if match.command('print'):
+ goal_id = int(match.args()[1])
+ logger.debug(f'Searching for goal {goal_id}...')
+ goal,tasks = Data().get_goal(goal_id)
+ results = pretty_print(goal, tasks)
+ return results
+
+
+ if match.command('print-vision'):
+ get_id = int(match.args()[1])
+ vision,goals = Data().get_vision(get_id)
+ results = pretty_print(vision, goals)
+ return results
+
+ if match.command('close-goal'):
+ goal_id = int(match.args()[1])
+ vision_id = Data().complete_goal(goal_id)
+ vision,goals = Data().get_vision(vision_id)
+ results = pretty_print(vision, goals)
+ return results
+
+ if match.command('close-vision'):
+ vision_id = int(match.args()[1])
+ Data().complete_vision(vision_id)
+ vision,goals = Data().get_vision(vision_id)
+ results = pretty_print(vision, goals)
+ return results
+
+
+ if match.command('done'):
+ task_id = int(match.args()[1])
+ goal_id = Data().complete_task(task_id)
+ goal,tasks = Data().get_goal(goal_id)
+ results = pretty_print(goal, tasks)
+ return results
+
+ if match.command('list-all'):
+ goals = Data().get_all_goals()
+ results = ['# Everything is...']
+ for goal in goals:
+ _,tasks = Data().get_goal(goal.id)
+ details = pretty_print(goal, tasks)
+ results += details
+ return results
+
+ if match.command('visions'):
+ items = Data().get_all_visions()
+ results = ['# Visions']
+ if not items:
+ results.append('-- No current visions --')
+
+ for item in items:
+ results.append(item)
+ return results
+
+
+ if match.command('set-vision'):
+ goal_id = int(match.args()[1])
+ vision_id = int(match.args()[2])
+ Data().assign_vision(goal_id=goal_id, vision_id=vision_id)
+ vision,goals = Data().get_vision(vision_id)
+ results = pretty_print(vision, goals)
+ return results
+
+ return [do_goal.__doc__]
+
+async def goal_bot(room, message, bot):
+ match = botlib.MessageMatch(room, message, bot, PREFIX)
+
+ if not match.is_not_from_this_bot() and match.prefix():
+ return
+
+ try:
+ if match.prefix():
+ result_lines = await do_goal(match)
+ for line in result_lines:
+ await bot.api.send_text_message(
+ room.room_id, str(line))
+ except Exception as ex:
+ return ['Ye cannot get ye flask.']
+
+
import logging
import os
from datetime import datetime as dt
+from decimal import Decimal
+from typing import List
from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
logger = logging.getLogger(__name__)
logging.basicConfig(filename=f'{MODULE}.log', encoding='utf-8', level=logging.DEBUG)
+class ItemStoreLink(SQLModel, table=True):
+ item_id: int | None = Field(default=None, foreign_key="item.id", primary_key=True)
+ store_id: int | None = Field(default=None, foreign_key="store.id", primary_key=True)
+
+
class Item(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
price: Decimal | None = Field(decimal_places=2)
- stores: list["Store"] = Relationship(back_populates="shopping_list")
+ stores: list["Store"] = Relationship(back_populates="shopping_list", link_model=ItemStoreLink)
purchase_date: dt | None = Field(default=None)
def __str__(self):
- return f'- [ ] {self.name} (#{self.id})'
+ return f'{self.name} (#{self.id})'
class Store(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
- shopping_list: list["Item"] = Relationship(back_populates="stores")
- purchase_history: list["PurchaseRecord"] = Relationship(back_populates="store")
+ shopping_list: list["Item"] = Relationship(back_populates="stores", link_model=ItemStoreLink)
+ # purchase_history: list["PurchaseRecord"] = Relationship(back_populates="store")
def __str__(self):
return f'{self.name} (#{self.id})'
-# May get used later...
-class PurchaseRecord(SQLModel, table=True):
- id: int | None = Field(default=None, primary_key=True)
- name: str = Field(default=None, unique=True)
- price: Decimal | None = Field(decimal_places=2)
- store: Store = Relationship(back_populates="purchase_history")
- purchase_date: dt | None = Field(default=None)
+# May get used later...
+#class PurchaseRecord(SQLModel, table=True):
+# id: int | None = Field(default=None, primary_key=True)
+# name: str = Field(default=None, unique=True)
+# price: Decimal | None = Field(decimal_places=2)
+# store: Store = Relationship(back_populates="purchase_history")
+# purchase_date: dt | None = Field(default=None)
+
+def _get_with_id(cls, item_id):
+ return select(cls).where(cls.id == item_id)
+
class Data():
def __init__(self, db_file=None):
if not db_file:
- db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), 'goals.db')
+ db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), f'{MODULE}.db')
self.file = db_file
self.engine = self.get_engine()
id = new.id
return id
- def _get_with_id(self, cls, item_id):
- with Session(self.engine) as session:
- cmd = select(cls).where(cls.id == item_id)
- return session.exec(cmd).first()
-
def add_store(self, name):
- return self.add_with_name(Store, name)
+ return self._add_with_name(Store, name)
def add_item(self, name):
- return self.add_with_name(Item, name)
+ return self._add_with_name(Item, name)
def assign_store(self, store_id, item_id):
with Session(self.engine) as session:
- store = self.get_with_id(Store, store_id)
- item = self.get_with_id(Item, item_id)
+ store = session.exec(_get_with_id(Store, store_id)).first()
+ item = session.exec(_get_with_id(Item, item_id)).first()
item.stores.append(store)
session.commit()
def purchase_item(self, item_id):
with Session(self.engine) as session:
- item = self.get_with_id(Item, item_id)
+ item = session.exec(_get_with_id(Item, item_id)).first()
item.purchase_date = dt.today()
session.commit()
def clear_list(self, store_id, clear_all=False):
with Session(self.engine) as session:
- store = self.get_with_id(Store, store_id)
+ store = session.exec(_get_with_id(Store, store_id)).first()
for item in store.items:
if clear_all or item.purchased_date:
item.remove()
return [f'No unsorted items!']
return items
- def get_shopping_list(self, store_id) -> List(str):
+ def get_shopping_list(self, store_id) -> List[str]:
results = []
with Session(self.engine) as session:
cmd = select(Store).where(Store.id == store_id)
items = store.shopping_list
if not items:
- results.append('-- Nothing to shop for at {store.name} --')
+ results.append(f'-- Nothing to shop for at {store.name} --')
for item in items:
if(item.purchase_date):
--- /dev/null
+from invoke import task
+
+from .db import Data
+
+@task
+def setup(c):
+ result = Data().setup()
+ print(result)
+
+@task
+def add_item(c, name):
+ result = Data().add_item(name)
+ print(result)
+
+@task
+def add_store(c, name):
+ result = Data().add_store(name)
+ print(result)
+
+
+@task
+def list(c, store_id):
+ items = Data().get_shopping_list(store_id)
+ for item in items:
+ print(item)
+
+
+@task
+def list_unsorted(c, store_id):
+ items = Data().list_unsorted()
+ for item in items:
+ print(item)
+
+@task
+def assign(c, store_id, item_id):
+ result = Data().assign_store(store_id, item_id)
+ print(result)
+
from matrix_shop_bot import tasks
ns = Collection()
-ns.add_collection(tasks)
+ns.add_collection(tasks, 'shop')
projects / matrix-shop-bot.git / commitdiff