From eb995294f17a9d4c5fd7e24e05570cbd7c840a28 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20Bu=C3=9Fmann?= Date: Fri, 14 Jun 2024 14:43:56 +0200 Subject: [PATCH] Log work in reclaim --- things2reclaim/__init__.py | 0 things2reclaim/{things2reclaim.py => main.py} | 97 ++++++++++++++++--- things2reclaim/reclaim_handler.py | 50 ++++++++-- things2reclaim/toggl_handler.py | 55 ++++++++++- things2reclaim/utils.py | 28 +++++- 5 files changed, 205 insertions(+), 25 deletions(-) create mode 100644 things2reclaim/__init__.py rename things2reclaim/{things2reclaim.py => main.py} (71%) diff --git a/things2reclaim/__init__.py b/things2reclaim/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/things2reclaim/things2reclaim.py b/things2reclaim/main.py similarity index 71% rename from things2reclaim/things2reclaim.py rename to things2reclaim/main.py index 6cad7ce..b0d38a7 100755 --- a/things2reclaim/things2reclaim.py +++ b/things2reclaim/main.py @@ -1,17 +1,18 @@ #!/opt/homebrew/Caskroom/miniconda/base/envs/things-automation/bin/python3 from datetime import datetime -from typing import Optional +from dateutil import tz +from typing import Optional, List, Dict import tomllib from pathlib import Path import sqlite3 import typer -from pytz import timezone from rich import print as rprint from rich.console import Console from rich.table import Table from rich.text import Text +from rich.prompt import Confirm from typing_extensions import Annotated import utils @@ -28,12 +29,15 @@ with open(CONFIG_PATH, "rb") as f: DATABASE_PATH = _config["database"]["path"] -app = typer.Typer(add_completion=False, no_args_is_help=True) +app = typer.Typer( + add_completion=False, no_args_is_help=True, pretty_exceptions_enable=False +) 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 @@ -44,6 +48,7 @@ def things_to_reclaim(things_task): if estimated_time is None: raise ValueError("EstimatedTime tag is required") estimated_time = utils.calculate_time_on_unit(estimated_time) + del tags["EstimatedTime"] params = { "name": things_handler.full_name(things_task), @@ -59,13 +64,13 @@ def things_to_reclaim(things_task): f"{things_task['start_date']} 08:00", "%Y-%m-%d %H:%M" ) if things_task.get("deadline"): - params["due_date"] = ( - datetime.strptime(f"{things_task['deadline']} 22:00", "%Y-%m-%d %H:%M"), + params["due_date"] = datetime.strptime( + f"{things_task['deadline']} 22:00", "%Y-%m-%d %H:%M" ) utils.map_tag_values(things_task, tags, params) - reclaim_handler.create_reaclaim_task(**params) + reclaim_handler.create_reaclaim_task_from_dict(params) @app.command("init") @@ -131,7 +136,7 @@ def list_reclaim_tasks(subject: Annotated[Optional[str], typer.Argument()] = Non reclaim_tasks = reclaim_handler.get_reclaim_tasks() if subject is not None: reclaim_tasks = reclaim_handler.filter_for_subject(subject, reclaim_tasks) - current_date = datetime.now().replace(tzinfo=timezone("UTC")) + current_date = datetime.now(tz.tzutc()) table = Table("Index", "Task", "Days left", title="Task list") for id, task in enumerate(reclaim_tasks): if current_date > task.due_date: @@ -153,11 +158,73 @@ def list_reclaim_tasks(subject: Annotated[Optional[str], typer.Argument()] = Non @app.command("start") def start_task( - task_name: Annotated[ - str, typer.Option(help="Task to start", autocompletion=complete_task_name) - ], + task_name_parts: Annotated[List[str], typer.Argument(help="Task to start")], ): - print(f"Starting task: {task_name}") + task_name = (" ").join(task_name_parts) + tasks: Dict[str, reclaim_handler.ReclaimTask] = { + task.name: task for task in reclaim_handler.get_reclaim_tasks() + } + if task_name not in tasks.keys(): + task = utils.get_closest_match(task_name, tasks) + if task is None: + utils.perror(f"No task with name {task_name} found") + return + else: + task = tasks[task_name] + + current_task = toggl_handler.get_current_time_entry() + if current_task is not None: + utils.perror("Toggl Track is already running") + return + + toggl_handler.start_task(task.name, reclaim_handler.get_project(task)) + reclaim_handler.start_task(task) + print(f"Started task {task.name}") + + +@app.command("stop") +def stop_task(): + current_task = toggl_handler.get_current_time_entry() + if current_task is None: + utils.perror("No task is currently tracked in toggl") + return + + stopped_task_name = current_task.description + + if stopped_task_name is None: + utils.perror("Current toggl task has no name") + return + + reclaim_dict = {task.name: task for task in reclaim_handler.get_reclaim_tasks()} + + if stopped_task_name in reclaim_dict.keys(): + stopped_task = reclaim_dict[stopped_task_name] + else: + stopped_task = utils.get_closest_match(stopped_task_name, reclaim_dict) + + 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() + + 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[") + 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}" + ) @app.command("stats") @@ -165,7 +232,7 @@ def show_task_stats(): """ Show task stats """ - current_date = datetime.now().replace(tzinfo=timezone("UTC")) + 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] @@ -208,7 +275,7 @@ def print_time_needed(): print("To many to-dos on list. Not all are scheduled") return last_task_date = tasks[-1].scheduled_start_date - today = datetime.now().replace(tzinfo=timezone("UTC")) + 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)" @@ -243,10 +310,10 @@ def sync_things_and_reclaim(verbose: bool = False): First updated all finished tasks in reclaim to completed in things Then upload all new tasks from things to reclaim """ - rprint("[bold white]Pulling from Reclaim[/bold white]") + utils.pinfo("Pulling from Reclaim") remove_finished_tasks_from_things() rprint("---------------------------------------------") - rprint("[bold white]Pushing to Reclaim[/bold white]") + utils.pinfo("Pushing to Reclaim") upload_things_to_reclaim(verbose=verbose) rprint("---------------------------------------------") diff --git a/things2reclaim/reclaim_handler.py b/things2reclaim/reclaim_handler.py index e99d24b..83bf985 100644 --- a/things2reclaim/reclaim_handler.py +++ b/things2reclaim/reclaim_handler.py @@ -1,9 +1,12 @@ -from typing import List -import tomllib +from datetime import datetime from pathlib import Path +from typing import List, Dict -from reclaim_sdk.models.task import ReclaimTask +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 CONFIG_PATH = Path("config/.reclaim.toml") @@ -17,7 +20,7 @@ RECLAIM_TOKEN = _config["reclaim_ai"]["token"] ReclaimClient(token=RECLAIM_TOKEN) -def get_reclaim_tasks(): +def get_reclaim_tasks() -> List[ReclaimTask]: return ReclaimTask.search() @@ -25,10 +28,45 @@ def filter_for_subject(subject, tasks): return [task for task in tasks if task.name.startswith(subject)] -def create_reaclaim_task(**params): - new_task = ReclaimTask(**params) +def get_project(task: ReclaimTask): + return task.name.split(" ")[0] + + +def start_task(task: ReclaimTask): + task.prioritize() + + +def create_reaclaim_task_from_dict(params: Dict): + new_task = ReclaimTask() + for key, value in params.items(): + setattr(new_task, key, value) new_task.save() +def log_work_for_task(task: ReclaimTask, start: datetime, end: datetime): + """ + start and end are in Europe/Berlin timezone + """ + utc = tz.tzutc() + if start.tzinfo is None: + raise ValueError("start is not timezone aware") + + if end.tzinfo is None: + raise ValueError("end is not timezone aware") + + if not task.is_scheduled: + raise RuntimeError("Task is not scheduled") + + last_event: ReclaimTaskEvent = task.events[-1] + + last_event.start = start.astimezone(utc) + last_event.end = end.astimezone(utc) + last_event._update() + + +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()] diff --git a/things2reclaim/toggl_handler.py b/things2reclaim/toggl_handler.py index 077da13..82e07ec 100644 --- a/things2reclaim/toggl_handler.py +++ b/things2reclaim/toggl_handler.py @@ -2,8 +2,12 @@ from toggl_python.entities import TimeEntry import toggl_python import tomllib from pathlib import Path +from dateutil import tz +import difflib from datetime import datetime, timedelta +from better_rich_prompts.prompt import ListPrompt + _config = {} @@ -25,6 +29,10 @@ time_entry_editor = toggl_python.WorkspaceTimeEntries( ) +def get_time_entry(id: int): + return toggl_python.TimeEntries(auth=auth).retrieve(id) + + def get_time_entries(since_days: int = 30): if since_days > 90: raise ValueError("since_days can't be more than 90 days") @@ -32,7 +40,7 @@ def get_time_entries(since_days: int = 30): return toggl_python.TimeEntries(auth=auth).list(since=time_stamp) -def get_current_time_entry(): +def get_current_time_entry() -> TimeEntry | None: time_entries = toggl_python.TimeEntries(auth=auth) time_entries.ADDITIONAL_METHODS = { "current": { @@ -44,16 +52,57 @@ def get_current_time_entry(): return time_entries.current() -def create_task_time_entry(description: str, project: str): +def get_tags(): + return toggl_python.Workspaces(auth=auth).tags(_id=workspace.id) + + +def get_approriate_tag(description: str) -> str | None: + tag_dict = {tag.name: tag for tag in get_tags()} + parts = description.replace("VL", "Vorlesung").split(" ") + possible_tags = set() + for part in parts: + possible_tags.update(difflib.get_close_matches(part, tag_dict.keys())) + + if not possible_tags: + print("Found no matching tags") + return + + possible_tags = list(possible_tags) + + if len(possible_tags) == 1: + return possible_tags[0] + else: + return ListPrompt.ask("Select the best fitting tag", possible_tags) + + +def create_task_time_entry( + description: str, project: str, start: datetime | None = None, duration: int = -1 +): + """ + duration is in seconds + """ if project not in project_dict.keys(): raise ValueError(f"{project} is not an active toggl project") + time_entry = TimeEntry( created_with="things-automation", wid=workspace.id, pid=project_dict[project].id, description=description, - duration=-1, + duration=duration, ) + + if start is not None: + if start.tzinfo is None: + raise ValueError("start has to be timezone aware") + else: + start = start.astimezone(tz.tzutc()) + time_entry.start = start + + tag = get_approriate_tag(description) + if tag: + time_entry.tags = [tag] + return time_entry diff --git a/things2reclaim/utils.py b/things2reclaim/utils.py index aa9447d..df15058 100644 --- a/things2reclaim/utils.py +++ b/things2reclaim/utils.py @@ -1,7 +1,11 @@ from datetime import datetime import re -from typing import Union, Dict +from typing import Union, Dict, Any, List +import difflib + +from rich import print as rprint import things_handler +from better_rich_prompts.prompt import ListPrompt regex = ( r"((\d+\.?\d*) (hours|hrs|hour|hr|h))? ?((\d+\.?\d*) (mins|min|minutes|minute|m))?" @@ -55,6 +59,28 @@ def map_tag_values( print(f"Tag {tag} not recognized") +def get_closest_match(name: str, candidates: Dict[str, Any]) -> Any | None: + possible_candidates: List[str] = difflib.get_close_matches(name, candidates.keys()) + if not possible_candidates: + return None + if len(possible_candidates) == 1: + return candidates[possible_candidates[0]] + else: + return candidates[ListPrompt.ask("Select a candidate", possible_candidates)] + + +def pinfo(msg: str): + rprint(f"[bold white]{msg}[/bold white]") + + +def pwarning(msg: str): + rprint(f"[bold yellow]Warning: {msg}[/bold yellow]") + + +def perror(msg: str): + rprint(f"[bold red]Error: {msg}[/bold red]") + + def generate_things_id_tag(things_task) -> str: return f"things_task:{things_task["uuid"]}"