import logging
import os
from datetime import datetime as dt
+from datetime import timedelta
+from typing import List
from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
+logger = logging.getLogger(__name__)
DATE_FORMAT = '%Y-%m-%d'
return dt.now().strftime(DATE_FORMAT)
-logger = logging.getLogger(__name__)
-logging.basicConfig(filename='goals.log', encoding='utf-8', level=logging.DEBUG)
-
-
class Vision(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
class Goal(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
+ start_date: dt | None = Field(default=dt.today())
complete_date: dt | None = Field(default=None)
vision_id: int | None = Field(default=None, foreign_key="vision.id")
notes: str | None = Field(default=None)
completed = self.complete_date.strftime(DATE_FORMAT)
return f'(Done {completed}) {self.name} (Goal {self.id})'
+
class Task(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
return f'{self.name} ({self.id})'
+class Win(SQLModel, table=True):
+ id: int | None = Field(default=None, primary_key=True)
+ goal: str = Field(default=None, unique=False)
+ vision: str | None = Field(default=None, unique=False)
+ complete_date: dt | None = Field(default=dt.today())
+ def __str__(self):
+ completed = self.complete_date.strftime(DATE_FORMAT)
+ return f'{self.goal} ({completed} {self.vision})'
+
def pretty_print(goal, tasks):
results = []
results.append(f' ## {goal}')
return results
+def _get_with_id(cls, item_id):
+ return select(cls).where(cls.id == item_id)
+
+def _get_with_name(cls, name):
+ return select(cls).where(func.lower(cls.name) == func.lower(name))
+
+
+def _get_goals_by_vision(goals) -> List[str]:
+ results = []
+ visions = {}
+
+ for goal in goals:
+ vision = str(goal.vision)
+ if not vision in visions:
+ visions[vision] = []
+ visions[vision].append(str(goal))
+ for vision in visions:
+ results.append('') # Make some space
+ results.append(f'## {vision}')
+ for item in visions[vision]:
+ results.append(' + ' + item)
+ return results
+
+
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', ''), 'goal.db')
self.file = db_file
self.engine = self.get_engine()
SQLModel.metadata.create_all(self.engine)
return 'Success'
+ def migrate(self):
+ with Session(self.engine) as session:
+ session.exec(text(''' alter table goal add start_date DATETIME; '''))
+ session.commit()
+ return 'Success'
+
def _set_pref_goal_id(self, goal_id):
with Session(self.engine) as session:
prefs = session.exec(select(PreferGoal)).all()
for pref in prefs:
session.delete(pref)
+ session.commit()
pref = PreferGoal()
pref.goal_id = goal_id
session.add(pref)
return pref.goal_id
+ def clear(self, goal_id, clear_all=False):
+ with Session(self.engine) as session:
+ goal = session.exec(_get_with_id(Goal, goal_id)).first()
+ for item in goal.tasks:
+ if clear_all or item.complete_date:
+ session.delete(item)
+ session.commit()
def get_all_visions(self):
with Session(self.engine) as session:
cmd = select(Vision).where(Vision.complete_date == None)
return session.exec(cmd).all()
- def print_ophan_tasks(self):
- pass
-
- def get_all_goals(self):
+ def get_all_goals(self) -> List[str]:
'''List all goals'''
+ results = []
with Session(self.engine) as session:
cmd = select(Goal).where(Goal.complete_date == None)
- return session.exec(cmd).all()
+ goals = session.exec(cmd).all()
+ results.append(_get_goals_by_vision(goals))
+ return results
+ def get_current_goals(self):
+ '''List current goals'''
+ with Session(self.engine) as session:
+ cmd = select(Goal).where(Goal.complete_date == None).where(Goal.start_date <= dt.today())
+ goals = session.exec(cmd).all()
+ logger.debug(f'Current goals: {goals}')
+ return _get_goals_by_vision(goals)
+
def get_done_goals(self):
'''List completed goals'''
with Session(self.engine) as session:
cmd = select(Goal).where(Goal.complete_date != None)
- return session.exec(cmd).all()
-
+ goals = session.exec(cmd).all()
+ return _get_goals_by_vision(goals)
def get_goal(self, goal_id):
self._set_pref_goal_id(goal_id)
def complete_goal(self, goal_id):
- self._set_pref_goal_id(goal_id)
vision_id = -1
with Session(self.engine) as session:
cmd = select(Goal).where(Goal.id == goal_id)
goal = session.exec(cmd).first()
goal.complete_date = dt.today()
vision_id = goal.vision_id
+
+ cmd = select(Vision).where(Vision.id == vision_id)
+ win = Win()
+ win.goal = goal.name
+
+ vision = session.exec(cmd).first()
+ if vision:
+ win.vision = vision.name
+
+ session.add(win)
session.commit()
return vision_id
+
+ def put_off_vision(self, vision_id):
+ with Session(self.engine) as session:
+ # vision = session.exec(cmd).first()
+ cmd = select(Goal).where(Goal.vision_id == vision_id)
+ goals = session.exec(cmd).all()
+ for goal in goals:
+ goal.start_date = dt.today() + timedelta(days=14)
+ session.commit()
+ return True
+
+ def start_goal(self, goal_id):
+ vision = None
+ with Session(self.engine) as session:
+ cmd = select(Goal).where(Goal.id == goal_id)
+ goal = session.exec(cmd).first()
+ goal.start_date = dt.today()
+ vision_id = goal.vision_id
+ session.commit()
+ if vision:
+ return (vision, vision.goals)
+ else:
+ return ('No Vision Assigned', [])
+
+
def rename_goal(self, goal_id, name):
self._set_pref_goal_id(goal_id)
vision_id = -1
def quick_add_task(self, name):
goal_id = self._get_pref_goal_id()
- return self.add_task(goal_id, name)
-
- def add_task(self, goal_id, name):
- self._set_pref_goal_id(goal_id)
- result = self.add_with_name(Task, name)
+ task_id = self.add_with_name(Task, name)
with Session(self.engine) as session:
- cmd = select(Task).where(Task.name == name)
+ cmd = select(Task).where(Task.id == task_id)
new = session.exec(cmd).first()
new.goal_id = goal_id
session.commit()
- return result
+ return goal_id
+
+ def remove_task(self, task_id):
+ task_name = ''
+ logger.debug(f'Removing task with id {task_id}')
+ with Session(self.engine) as session:
+ cmd = select(Task).where(Task.id == int(task_id))
+ to_remove = session.exec(cmd).first()
+ if not to_remove:
+ return f'Unable to find task with id {task_id}'
+
+ logger.debug(f'Removing task with name {to_remove.name}')
+ task_name = to_remove.name
+ session.delete(to_remove)
+ session.commit()
+ return f'Removed task "{task_name}"'
def assign_vision(self, goal_id, vision_id):
'''Goals may move between visions'''
goal = session.exec(cmd).first()
goal.vision_id = vision_id
session.commit()
+ return True
def visionless(self):
'''Get goals with no vision'''
with Session(self.engine) as session:
- cmd = select(Goal).where(Goal.vision_id == None)
+ cmd = select(Goal).where(Goal.vision_id == None).where(Goal.complete_date == None)
return session.exec(cmd).all()
'''Notes print with the goal.'''
pass
+
+ def wins(self):
+ results = []
+ with Session(self.engine) as session:
+ results = session.exec(select(Win)).all()
+ return results
+
# Someday: Notes on each
projects / matrix-goal-bot.git / blobdiff