diff --git a/things2reclaim/deadline_status.py b/things2reclaim/deadline_status.py new file mode 100644 index 0000000..28aa661 --- /dev/null +++ b/things2reclaim/deadline_status.py @@ -0,0 +1,6 @@ +from enum import Enum + +class DeadlineStatus(Enum): + FINE = 1 + OVERDUE = 2 + NONE = 3 diff --git a/things2reclaim/main.py b/things2reclaim/main.py index 784a87b..1de61bf 100755 --- a/things2reclaim/main.py +++ b/things2reclaim/main.py @@ -3,16 +3,9 @@ import sqlite3 from datetime import datetime from pathlib import Path -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Union import tomllib -import reclaim_handler -import things_handler -import toggl_handler -import utils -from database_handler import UploadedTasksDB - - from dateutil import tz from rich import print as rprint from rich.console import Console @@ -22,6 +15,12 @@ from rich.text import Text from typing_extensions import Annotated import typer +import reclaim_handler +from deadline_status import DeadlineStatus +import things_handler +import toggl_handler +import utils +from database_handler import UploadedTasksDB CONFIG_PATH = Path("config/.things2reclaim.toml") @@ -37,13 +36,6 @@ app = typer.Typer( console = Console() -def complete_task_name(incomplete: str): - for name in [task.name for task in reclaim_handler.get_reclaim_tasks()]: - print(name) - if name.startswith(incomplete): - yield name - - def things_to_reclaim(things_task): tags = things_handler.get_task_tags(things_task) estimated_time = tags.get("EstimatedTime") @@ -75,6 +67,21 @@ def things_to_reclaim(things_task): reclaim_handler.create_reaclaim_task_from_dict(params) +def finish_task(task : Union[reclaim_handler.ReclaimTask, str]): + """ + Finish task + """ + if isinstance(task, str): + things_id = task + elif isinstance(task, reclaim_handler.ReclaimTask): + things_id = reclaim_handler.get_things_id(task) + reclaim_handler.finish_task(task) + + things_handler.complete(things_id) + with UploadedTasksDB(DATABASE_PATH) as db: + db.remove_uploaded_task(things_id) + + @app.command("init") def initialize_uploaded_database(verbose: bool = False): """ @@ -111,8 +118,7 @@ def upload_things_to_reclaim(verbose: bool = False): Upload things tasks to reclaim """ projects = things_handler.extract_uni_projects() - reclaim_task_names = [ - task.name for task in reclaim_handler.get_reclaim_tasks()] + reclaim_task_names = reclaim_handler.get_reclaim_task_names() tasks_uploaded = 0 with UploadedTasksDB(DATABASE_PATH) as db: for project in projects: @@ -150,20 +156,23 @@ def list_reclaim_tasks(subject: Annotated[Optional[str], current_date = datetime.now(tz.tzutc()) table = Table("Index", "Task", "Days left", title="Task list") for index, task in enumerate(reclaim_tasks): - if current_date > task.due_date: - days_behind = (current_date - task.due_date).days - table.add_row( - f"({index + 1})", - task.name, - Text(f"{days_behind} days overdue", style="bold red"), - ) - else: - days_left = (task.due_date - current_date).days - table.add_row( - f"({index + 1})", - task.name, - Text(f"{days_left} days left", style="bold white"), - ) + due_date = task.due_date + if due_date is None: + date_str = Text() + else: + if current_date > due_date: + days_behind = (current_date - due_date).days + date_str = Text(f"{days_behind} days overdue") + date_str.stylize("bold red") + else: + days_left = (due_date - current_date).days + date_str = Text(f"{days_left} days left") + date_str.stylize("bold white") + table.add_row( + f"({index + 1})", + task.name, + date_str) + console.print(table) @@ -216,29 +225,30 @@ def stop_task(): if stopped_task is None: utils.perror(f"{stopped_task} not found in reclaim") return + stop_time = datetime.now(tz.tzutc()) if callable(current_task.start): current_task.start = current_task.start() toggl_handler.stop_current_task() + is_task_finished = Confirm.ask("Is task finished?", default=False) + if stopped_task.is_scheduled: reclaim_handler.log_work_for_task( stopped_task, current_task.start, stop_time) - - is_task_finished = Confirm.ask("Is task finished?", default=False) - if is_task_finished: - reclaim_handler.finish_task(stopped_task) - rprint(f"Finished {stopped_task.name}") else: - utils.pwarning("Work could not be logged in reclaim[") + utils.pwarning("Work could not be logged in reclaim!") + + if is_task_finished: + finish_task(stopped_task) + rprint(f"Finished {stopped_task.name}") + time_format = "%H:%M" local_zone = tz.gettz() rprint( - f"Logged work from \ - {current_task.start.astimezone(local_zone).strftime(time_format)} to \ - {stop_time.astimezone(local_zone).strftime(time_format)} \ - for {stopped_task.name}" + (f"Logged work from {current_task.start.astimezone(local_zone).strftime(time_format)} " + f"to {stop_time.astimezone(local_zone).strftime(time_format)} for {stopped_task.name}") ) @@ -249,10 +259,9 @@ def show_task_stats(): """ current_date = datetime.now(tz.tzutc()) reclaim_tasks = reclaim_handler.get_reclaim_tasks() - tasks_fine = [ - task for task in reclaim_tasks if task.due_date >= current_date] - tasks_overdue = [ - task for task in reclaim_tasks if task.due_date < current_date] + + tasks_fine = reclaim_handler.filter_for_deadline_status(current_date, DeadlineStatus.FINE, reclaim_tasks) + tasks_overdue = reclaim_handler.filter_for_deadline_status(current_date, DeadlineStatus.OVERDUE, reclaim_tasks) fine_per_course = ["Fine"] overdue_per_course = ["Overdue"] @@ -283,24 +292,26 @@ def print_time_needed(): time_needed = 0 for task in tasks: - time_needed += task.duration + task_duration = task.duration + if task_duration: + time_needed += task_duration print(f"Time needed to complete {len(tasks)} Tasks: {time_needed} hrs") print(f"Average time needed to complete a Task: { time_needed/len(tasks):.2f} hrs") + + # sort unscheduled tasks to the end of the list + tasks.sort(key=lambda x: x.scheduled_start_date if x.scheduled_start_date is not None else float('inf')) - try: - tasks.sort(key=lambda x: x.scheduled_start_date) - except TypeError: - print("To many to-dos on list. Not all are scheduled") - return last_task_date = tasks[-1].scheduled_start_date - today = datetime.now(tz.tzutc()) - - print( - f"""Last task is scheduled for {last_task_date.strftime('%d.%m.%Y')} - ({last_task_date - today} till completion)""" - ) + if last_task_date is None: # last task on todo list is not scheduled + print("Too many tasks on todo list. Not all are scheduled.") + else: + today = datetime.now(tz.tzutc()) + print( + f"""Last task is scheduled for {last_task_date.strftime('%d.%m.%Y')} + ({last_task_date - today} till completion)""" + ) @app.command("finished") @@ -310,21 +321,29 @@ def remove_finished_tasks_from_things(): """ reclaim_things_uuids = reclaim_handler.get_reclaim_things_ids() finished_someting = False - with UploadedTasksDB(DATABASE_PATH) as db: - for task in things_handler.get_all_uploaded_things_tasks(): - if task["uuid"] not in reclaim_things_uuids: - finished_someting = True - print( - f"Found completed task: { - things_handler.full_name(things_task=task)}" - ) - things_handler.complete(task["uuid"]) - db.remove_uploaded_task(task["uuid"]) + for task in things_handler.get_all_uploaded_things_tasks(): + if task["uuid"] not in reclaim_things_uuids: + finished_someting = True + print( + f"Found completed task: { + things_handler.full_name(things_task=task)}" + ) + finish_task(task["uuid"]) if not finished_someting: print("Reclaim and Things are synced") +@app.command("tracking") +def sync_toggl_reclaim_tracking(): + since_days = 0 + reclaim_handler.get_reclaim_tasks() + toggl_time_entries = toggl_handler.get_time_entries_since(since_days = since_days) # end date is inclusive + reclaim_time_entries = reclaim_handler.get_task_events_since(since_days = since_days) # end date is inclusive + rprint(reclaim_time_entries) + + + @app.command("sync") def sync_things_and_reclaim(verbose: bool = False): """ diff --git a/things2reclaim/reclaim_handler.py b/things2reclaim/reclaim_handler.py index 042dea3..6d753d3 100644 --- a/things2reclaim/reclaim_handler.py +++ b/things2reclaim/reclaim_handler.py @@ -1,13 +1,17 @@ from datetime import datetime, date, timedelta from pathlib import Path -from typing import List, Dict +from typing import List, Dict, Pattern, Optional +import re +import emoji import tomllib from dateutil import tz from reclaim_sdk.client import ReclaimClient from reclaim_sdk.models.task import ReclaimTask from reclaim_sdk.models.task_event import ReclaimTaskEvent +from deadline_status import DeadlineStatus + CONFIG_PATH = Path("config/.reclaim.toml") _config = {} @@ -17,6 +21,10 @@ with open(CONFIG_PATH, "rb") as f: RECLAIM_TOKEN = _config["reclaim_ai"]["token"] +THINGS_ID_PATTERN = "things_task:[a-zA-z0-9]+" + +things_id_pattern : Pattern[str] = re.compile(THINGS_ID_PATTERN) + ReclaimClient(token=RECLAIM_TOKEN) @@ -24,18 +32,43 @@ def get_reclaim_tasks() -> List[ReclaimTask]: return ReclaimTask.search() +def get_reclaim_task_names(tasks : Optional[List[ReclaimTask]] = None): + if not tasks: + tasks = get_reclaim_tasks() + return [task.name for task in tasks] + + def filter_for_subject(subject, tasks): return [task for task in tasks if task.name.startswith(subject)] +def filter_for_deadline_status(cur_date : datetime, deadline_status : DeadlineStatus, tasks : List[ReclaimTask]): + # [task for task in reclaim_tasks if task.due_date >= current_date] + match deadline_status: + case DeadlineStatus.FINE: + return [task for task in tasks if (task.due_date and task.due_date >= cur_date)] + case DeadlineStatus.OVERDUE: + return [task for task in tasks if (task.due_date and task.due_date < cur_date)] + case DeadlineStatus.NONE: + return [task for task in tasks if task.due_date is None] + + def get_project(task: ReclaimTask): return task.name.split(" ")[0] -def get_events_since(since_days: int = 1): +def get_clean_time_entry_name(name : str): + return emoji.replace_emoji(name).lstrip() + +def get_task_events_since(since_days: int = 0) -> List[ReclaimTaskEvent]: date_now = datetime.now(tz.tzutc()).date() date_since = date_now - timedelta(days=since_days) - return ReclaimTaskEvent.search(date_since, date_now) + date_end = date_now + timedelta(days = 1) # end date is exclusive + events = ReclaimTaskEvent.search(date_since, date_end) + task_names = get_reclaim_task_names() + print(task_names) + [print(get_clean_time_entry_name(event.name)) for event in events] + return [event for event in events if get_clean_time_entry_name(event.name) in task_names] def get_events_date_range(from_date: date, to_date: date): @@ -61,10 +94,10 @@ def log_work_for_task(task: ReclaimTask, start: datetime, end: datetime): raise ValueError("end is not timezone aware") if not task.is_scheduled: - raise RuntimeError("Task is not scheduled") + raise ValueError("Task is not scheduled") if not task.events: - raise RuntimeError("Event list is empty") + raise ValueError("Event list is empty") last_event: ReclaimTaskEvent = task.events[-1] @@ -77,5 +110,13 @@ def finish_task(task: ReclaimTask): task.mark_complete() -def get_reclaim_things_ids() -> List[str]: - return [task.description.split(":")[1].strip() for task in ReclaimTask.search()] +def get_things_id(task: ReclaimTask): + things_id_match = things_id_pattern.match(task.description) + if things_id_match is None: + raise ValueError(f"things id could not be found in description of reclaim task {task.name}") + things_id_tag : str = things_id_match.group() + + return things_id_tag.split(":")[1] + +def get_reclaim_things_ids() -> List[str]: + return [get_things_id(task) for task in ReclaimTask.search()] diff --git a/things2reclaim/things_handler.py b/things2reclaim/things_handler.py index e76e934..9ebf35a 100644 --- a/things2reclaim/things_handler.py +++ b/things2reclaim/things_handler.py @@ -20,11 +20,11 @@ def extract_uni_projects(): return things.projects(area=uni_area["uuid"]) -def get_task(task_id: int): +def get_task(task_id: str): return things.get(task_id) -def complete(task_id: int): +def complete(task_id: str): things.complete(task_id) diff --git a/things2reclaim/toggl_handler.py b/things2reclaim/toggl_handler.py index 5541daa..e7c346e 100644 --- a/things2reclaim/toggl_handler.py +++ b/things2reclaim/toggl_handler.py @@ -1,5 +1,5 @@ import difflib -from datetime import datetime, timedelta, date +from datetime import datetime, timedelta, date, time from pathlib import Path import tomllib @@ -39,9 +39,13 @@ def get_time_entries_date_range(from_date: date, to_date: date): def get_time_entries_since(since_days: int = 30): + """ + get time entries since days at midnight + """ if since_days > 90: raise ValueError("since_days can't be more than 90 days") - time_stamp = int((datetime.now() - timedelta(days=since_days)).timestamp()) + midnight = datetime.combine(datetime.now(), time.min) + time_stamp = int((midnight - timedelta(days=since_days)).timestamp()) return toggl_python.TimeEntries(auth=auth).list(since=time_stamp) diff --git a/things2reclaim/utils.py b/things2reclaim/utils.py index 90922e5..6e47237 100644 --- a/things2reclaim/utils.py +++ b/things2reclaim/utils.py @@ -1,6 +1,6 @@ from datetime import datetime import re -from typing import Union, Dict, Any, List +from typing import Union, Dict, TypeVar, List import difflib from better_rich_prompts.prompt import ListPrompt @@ -13,6 +13,7 @@ TIME_PATTERN = ( ) pattern = re.compile(TIME_PATTERN) +T = TypeVar("T") # generic type def calculate_time_on_unit(tag_value: str) -> float: # This is a regex to match time in the format of 1h 30m @@ -60,8 +61,9 @@ def map_tag_values( print(f"Tag {tag} not recognized") -def get_closest_match(name: str, candidates: Dict[str, Any]) -> Any | None: +def get_closest_match(name: str, candidates: Dict[str, T]) -> T | None: possible_candidates: List[str] = difflib.get_close_matches(name, candidates.keys()) + if not possible_candidates: return None if len(possible_candidates) == 1: