diff --git a/things2reclaim/main.py b/things2reclaim/main.py index a0c9b56..ab664bc 100755 --- a/things2reclaim/main.py +++ b/things2reclaim/main.py @@ -20,6 +20,7 @@ import reclaim_handler from deadline_status import DeadlineStatus import things_handler import toggl_handler +import task_scheduler_handler import utils from database_handler import UploadedTasksDB @@ -34,8 +35,7 @@ DATABASE_PATH = utils.get_project_root() / _config["database"]["path"] app = typer.Typer(add_completion=False, no_args_is_help=True) console = Console() - -def things_to_reclaim(things_task): +def generate_params_dict(things_task): tags = things_handler.get_task_tags(things_task) estimated_time = tags.get("EstimatedTime") if estimated_time is None: @@ -62,9 +62,17 @@ def things_to_reclaim(things_task): ) utils.map_tag_values(things_task, tags, params) + return params + +def things_to_reclaim(things_task): + params = generate_params_dict(things_task) reclaim_handler.create_reaclaim_task_from_dict(params) +def things_to_task(things_task): + params = generate_params_dict(things_task) + task_scheduler_handler.create_task_from_dict(params) + def finish_task(task: Union[reclaim_handler.ReclaimTask, str]): """ @@ -452,6 +460,17 @@ def sync_things_and_reclaim(dry_run: bool = False): upload_things_to_reclaim(dry_run) rprint("---------------------------------------------") +@app.command("upload_to_scheduler") +def upload_to_scheduler(dry_run: bool = False): + """ + Upload things tasks to task-scheduler + """ + tasks = things_handler.get_all_things_tasks() + for task in tasks: + print(f"Creating task {things_handler.full_name(task)} in Task Scheduler") + if not dry_run: + things_to_task(task) + print(f"Uploaded {len(tasks)} task{'s' if len(tasks) > 1 else ''}") if __name__ == "__main__": app() diff --git a/things2reclaim/task_scheduler_handler.py b/things2reclaim/task_scheduler_handler.py new file mode 100644 index 0000000..03c51c8 --- /dev/null +++ b/things2reclaim/task_scheduler_handler.py @@ -0,0 +1,40 @@ +from typing import Dict, List + +from task_scheduler_client.configuration import Configuration +from task_scheduler_client.api.default_api import DefaultApi +from task_scheduler_client.api_client import ApiClient +from task_scheduler_client.models.task import Task + +client = ApiClient(Configuration(host="http://localhost:8000")) +api = DefaultApi(client) + +def get_tasks() -> List[Task]: + return api.get_tasks_tasks_get() + +def remap_dict_keys(params: Dict): + key_mapping = {"min_work_duration": "min_time", "max_work_duration": "max_work", "duration": "estimated_time", "due_date": "deadline", "description": "things_id"} + return {key_mapping.get(k,k): v for k, v in params.items()} + +def remove_keys(params: Dict): + keys_to_remove = {"tags"} + return {k: v for k, v in params.items() if k not in keys_to_remove} + +def change_value(params: Dict): + keys_to_change = {"things_id": "things_task:"} + return {k: (v if k not in keys_to_change else v.removeprefix(keys_to_change[k])) for k, v in params.items()} + +def param_dict_to_task(params: Dict): + remapped_keys = remap_dict_keys(params) + stripped_keys = remove_keys(remapped_keys) + return change_value(stripped_keys) + +def create_task(task: Task): + api.create_task_tasks_post(task) + +def create_task_from_dict(params: Dict): + params = param_dict_to_task(params) + task = Task.from_dict(params) + if task is None: + raise ValueError + api.create_task_tasks_post(task) +