mirrors
/
autodoist
mirror of https://github.com/Hoffelhas/autodoist
1
0
Fork 1
Code Issues Releases Activity

Basic labelling functionality working again. Still need to fix metadata storage

pull/30/head
Hoffelhas 2023-01-02 16:28:16 +01:00
parent 95a71524cd
commit 2625547002
3 changed files with 276 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Procfile
View File

@ -1 +0,0 @@
nextaction: python nextaction.py -a $TODOIST_API_KEY -l $TODOIST_NEXT_ACTION_LABEL $DEBUG --nocache

514
autodoist.py
View File

@ -8,6 +8,8 @@ import argparse
import logging
from datetime import datetime, timedelta
import time
import sqlite3
from sqlite3 import Error
# Makes --help text wider
@ -79,9 +81,9 @@ def verify_label_existance(api, label_name, prompt_mode):
label = [x for x in labels if x.name == label_name]
if len(label) > 0:
label_id = label[0].id
next_action_label = label[0].id
logging.debug('Label \'%s\' found as label id %d',
label_name, label_id)
label_name, next_action_label)
else:
# Create a new label in Todoist
logging.info(
@ -101,15 +103,14 @@ def verify_label_existance(api, label_name, prompt_mode):
labels = api.get_labels()
label = [x for x in labels if x.name == label_name]
label_id = label[0].id
next_action_label = label[0].id
logging.info("Label '{}' has been created!".format(label_name))
else:
logging.info('Exiting Autodoist.')
exit(1)
return label_id
return 0
# Initialisation of Autodoist
def initialise(args):
@ -171,11 +172,7 @@ def initialise(args):
if args.label is not None:
# Verify that the next action label exists; ask user if it needs to be created
label_id = verify_label_existance(api, args.label, 1)
else:
# Label functionality not needed
label_id = None
verify_label_existance(api, args.label, 1)
# If regeneration mode is used, verify labels
if args.regeneration is not None:
@ -188,7 +185,7 @@ def initialise(args):
# Label functionality not needed
regen_labels_id = [None, None, None]
return api, label_id, regen_labels_id
return api
# Check for Autodoist update
@ -232,19 +229,20 @@ def check_name(args, name):
current_type = 'parallel'
elif name[-len_suffix[1]:] == args.ss_suffix:
current_type = 'sequential'
elif name[-len_suffix[1]:] == args.ps_suffix:
elif name[-len_suffix[2]:] == args.ps_suffix:
current_type = 'p-s'
elif name[-len_suffix[1]:] == args.sp_suffix:
elif name[-len_suffix[3]:] == args.sp_suffix:
current_type = 's-p'
# Workaround for section names, which don't allow / symbol.
elif args.ps_suffix == '/-' and name[-2:] == '_-':
current_type = 'p-s'
# Workaround for section names, which don't allow / symbol.
elif args.sp_suffix == '-/' and name[-2:] == '-_':
current_type = 's-p'
# Workaround for section names, which don't allow / symbol.
elif args.pp_suffix == '//' and name[-1:] == '_':
current_type = 'parallel'
#TODO: Remove below workarounds if standard notation is changing. Just messy and no longer needed.
# # Workaround for section names, which don't allow '/' symbol.
# elif args.ps_suffix == '/-' and name[-2:] == '_-':
# current_type = 'p-s'
# # Workaround for section names, which don't allow '/' symbol.
# elif args.sp_suffix == '-/' and name[-2:] == '-_':
# current_type = 's-p'
# # Workaround for section names, which don't allow '/' symbol.
# elif args.pp_suffix == '//' and name[-1:] == '_':
# current_type = 'parallel'
else:
current_type = None
@ -253,42 +251,45 @@ def check_name(args, name):
# Scan the end of a name to find what type it is
def get_type(args, object, key):
def get_type(args, model, key):
object_name = ''
model_name = ''
try:
old_type = object[key]
old_type = model[key] #TODO: METADATA: this information used to be part of the metadata, needs to be retreived from own database
except:
# logging.debug('No defined project_type: %s' % str(e))
old_type = None
try:
object_name = object['name'].strip()
except:
try:
object_name = object['content'].strip()
model_name = model.name.strip()
except:
#TODO: Old support for legacy tag in v1 API, can likely be removed since moving to v2.
# try:
#
# object_name = object['content'].strip()
# except:
# pass
pass
current_type = check_name(args, object_name)
current_type = check_name(args, model_name)
# Check if project type changed with respect to previous run
if old_type == current_type:
type_changed = 0
else:
type_changed = 1
object[key] = current_type
# model.key = current_type #TODO: METADATA: this information used to be part of the metadata, needs to be retreived from own database
return current_type, type_changed
# Determine a project type
def get_project_type(args, project_object):
def get_project_type(args, project_model):
"""Identifies how a project should be handled."""
project_type, project_type_changed = get_type(
args, project_object, 'project_type')
args, project_model, 'project_type')
return project_type, project_type_changed
@ -306,65 +307,67 @@ def get_section_type(args, section_object):
return section_type, section_type_changed
# Determine an item type
# Determine an task type
def get_item_type(args, item, project_type):
"""Identifies how an item with sub items should be handled."""
def get_task_type(args, task, project_type):
"""Identifies how a task with sub tasks should be handled."""
if project_type is None and item['parent_id'] != 0:
if project_type is None and task.parent_id != 0:
try:
item_type = item['parent_type']
item_type_changed = 1
item['item_type'] = item_type
task_type = task.parent_type #TODO: METADATA
task_type_changed = 1
task.task_type = task_type
except:
item_type, item_type_changed = get_type(args, item, 'item_type')
task_type, task_type_changed = get_type(args, task, 'task_type') #TODO: METADATA
else:
item_type, item_type_changed = get_type(args, item, 'item_type')
task_type, task_type_changed = get_type(args, task, 'task_type') #TODO: METADATA
return item_type, item_type_changed
return task_type, task_type_changed
# Logic to add a label to an item
# Logic to track addition of a label to a task
def add_label(item, label, overview_item_ids, overview_item_labels):
if label not in item['labels']:
labels = item['labels']
logging.debug('Updating \'%s\' with label', item['content'])
def add_label(task, label, overview_task_ids, overview_task_labels):
if label not in task.labels:
labels = task.labels # Copy other existing labels
logging.debug('Updating \'%s\' with label', task.content)
labels.append(label)
try:
overview_item_ids[str(item['id'])] += 1
overview_task_ids[task.id] += 1
except:
overview_item_ids[str(item['id'])] = 1
overview_item_labels[str(item['id'])] = labels
overview_task_ids[task.id] = 1
overview_task_labels[task.id] = labels
# Logic to remove a label from an item
# Logic to track removal of a label from a task
def remove_label(item, label, overview_item_ids, overview_item_labels):
if label in item['labels']:
labels = item['labels']
logging.debug('Removing \'%s\' of its label', item['content'])
def remove_label(task, label, overview_task_ids, overview_task_labels):
if label in task.labels:
labels = task.labels
logging.debug('Removing \'%s\' of its label', task.content)
labels.remove(label)
try:
overview_item_ids[str(item['id'])] -= 1
overview_task_ids[task.id] -= 1
except:
overview_item_ids[str(item['id'])] = -1
overview_item_labels[str(item['id'])] = labels
overview_task_ids[task.id] = -1
overview_task_labels[task.id] = labels
# Ensure labels are only issued once per item
# Ensure label updates are only issued once per task
def update_labels(api, label_id, overview_item_ids, overview_item_labels):
def update_labels(api, overview_task_ids, overview_task_labels):
filtered_overview_ids = [
k for k, v in overview_item_ids.items() if v != 0]
for item_id in filtered_overview_ids:
labels = overview_item_labels[item_id]
api.items.update(item_id, labels=labels)
k for k, v in overview_task_ids.items() if v != 0]
for task_id in filtered_overview_ids:
labels = overview_task_labels[task_id]
api.update_task(task_id=task_id, labels=labels)
# To handle items which have no sections
return filtered_overview_ids
# To handle tasks which have no sections
def create_none_section():
@ -390,7 +393,7 @@ def check_header(level):
except:
try:
# Current structure
content = level['content']
content = level.content
method = 2
except:
pass
@ -414,38 +417,54 @@ def check_header(level):
return header_all_in_level, unheader_all_in_level
# Logic for applying and removing headers
def modify_headers(task, child_tasks, header_all_in_p, unheader_all_in_p, header_all_in_s, unheader_all_in_s, header_all_in_t, unheader_all_in_t):
if any([header_all_in_p, header_all_in_s, header_all_in_t]):
if task.content[0] != '*':
task.update(content='* ' + task.content)
for ci in child_tasks:
if not ci.content.startswith('*'):
ci.update(content='* ' + ci.content)
if any([unheader_all_in_p, unheader_all_in_s]):
if task.content[0] == '*':
task.update(content=task.content[2:])
if unheader_all_in_t:
[ci.update(content=ci.content[2:])
for ci in child_tasks]
# Check regen mode based on label name
def check_regen_mode(api, item, regen_labels_id):
labels = item['labels']
labels = item.labels
overlap = set(labels) & set(regen_labels_id)
overlap = [val for val in overlap]
if len(overlap) > 1:
logging.warning(
'Multiple regeneration labels used! Please pick only one for item: "{}".'.format(item['content']))
'Multiple regeneration labels used! Please pick only one for item: "{}".'.format(item.content))
return None
try:
regen_label_id = overlap[0]
regen_next_action_label = overlap[0]
except:
logging.debug(
'No regeneration label for item: %s' % item['content'])
regen_label_id = [0]
'No regeneration label for item: %s' % item.content)
regen_next_action_label = [0]
if regen_label_id == regen_labels_id[0]:
if regen_next_action_label == regen_labels_id[0]:
return 0
elif regen_label_id == regen_labels_id[1]:
elif regen_next_action_label == regen_labels_id[1]:
return 1
elif regen_label_id == regen_labels_id[2]:
elif regen_next_action_label == regen_labels_id[2]:
return 2
else:
# label_name = api.labels.get_by_id(regen_label_id)['name']
# label_name = api.labels.get_by_id(regen_next_action_label)['name']
# logging.debug(
# 'No regeneration label for item: %s' % item['content'])
# 'No regeneration label for item: %s' % item.content)
return None
@ -472,10 +491,10 @@ def run_recurring_lists_logic(args, api, item, child_items, child_items_all, reg
if regen_mode is None:
regen_mode = args.regeneration
logging.debug('Using general recurring mode \'%s\' for item: %s',
regen_mode, item['content'])
regen_mode, item.content)
else:
logging.debug('Using recurring label \'%s\' for item: %s',
regen_mode, item['content'])
regen_mode, item.content)
# Apply tags based on mode
give_regen_tag = 0
@ -519,7 +538,7 @@ def run_recurring_lists_logic(args, api, item, child_items, child_items_all, reg
# Only apply if overdue and if it's a daily recurring tasks
if days_overdue >= 1 and days_difference == 1:
# Find curreny date in string format
# Find current date in string format
today_str = [str(x) for x in [
today.year, today.month, today.day]]
if len(today_str[1]) == 1:
@ -540,13 +559,13 @@ def run_recurring_lists_logic(args, api, item, child_items, child_items_all, reg
except:
# If date has never been saved before, create a new entry
logging.debug(
'New recurring task detected: %s' % item['content'])
'New recurring task detected: %s' % item.content)
item['date_old'] = item['due']['date'][:10]
api.items.update(item['id'])
except:
# logging.debug(
# 'Parent not recurring: %s' % item['content'])
# 'Parent not recurring: %s' % item.content)
pass
if args.regeneration is not None and item['parent_id'] != 0:
@ -561,17 +580,17 @@ def run_recurring_lists_logic(args, api, item, child_items, child_items_all, reg
child_item['r_tag'] = 1
except:
# logging.debug('Child not recurring: %s' %
# item['content'])
# item.content)
pass
# Contains all main autodoist functionalities
def autodoist_magic(args, api, label_id, regen_labels_id):
def autodoist_magic(args, api, next_action_label, regen_labels_id):
# Preallocate dictionaries
overview_item_ids = {}
overview_item_labels = {}
overview_task_ids = {}
overview_task_labels = {}
try:
projects = api.get_projects()
@ -586,25 +605,29 @@ def autodoist_magic(args, api, label_id, regen_labels_id):
# Check if we need to (un)header entire project
header_all_in_p, unheader_all_in_p = check_header(project)
if label_id is not None:
# Get project type
if next_action_label is not None:
project_type, project_type_changed = get_project_type(
args, project)
if project_type is not None:
logging.debug('Identified \'%s\' as %s type',
project['name'], project_type)
project.name, project_type)
# Get all items for the project
# Get all tasks for the project
try:
project_tasks = api.get_tasks(project_id = project.id)
except Exception as error:
print(error)
# Run for both none-sectioned and sectioned items
# Run for both non-sectioned and sectioned tasks
# Get completed tasks: get(api._session, endpoint, api._token, '0')['items']
# Get completed tasks:
# endpoint = 'https://api.todoist.com/sync/v9/completed/get_all'
# get(api._session, endpoint, api._token, '0')['items']
# $ curl https://api.todoist.com/sync/v9/sync-H "Authorization: Bearer e2f750b64e8fc06ae14383d5e15ea0792a2c1bf3" -d commands='[ {"type": "item_add", "temp_id": "63f7ed23-a038-46b5-b2c9-4abda9097ffa", "uuid": "997d4b43-55f1-48a9-9e66-de5785dfd69b", "args": {"content": "Buy Milk", "project_id": "2203306141","labels": ["Food", "Shopping"]}}]'
# for s in [0, 1]: # TEMP SECTION ONLY
# for s in [0, 1]: # TODO: TEMPORARELY SKIP SECTIONLESS TASKS
for s in [1]:
if s == 0:
sections = [create_none_section()] # TODO: Rewrite
@ -627,208 +650,199 @@ def autodoist_magic(args, api, label_id, regen_labels_id):
args, section)
if section_type is not None:
logging.debug('Identified \'%s\' as %s type',
section['name'], section_type)
section.name, section_type)
# Get all items for the section
# Get all tasks for the section
tasks = [x for x in project_tasks if x.section_id
== section.id]
# Change top parents_id in order to numerically sort later on
# Change top tasks parents_id from 'None' to '0' in order to numerically sort later on
for task in tasks:
if not task.parent_id:
task.parent_id = 0
# Sort by parent_id and filter for completable items
items = sorted(items, key=lambda x: (
x['parent_id'], x['child_order']))
# Sort by parent_id and child order
# In the past, Todoist used to screw up the tasks orders, so originally I processed parentless tasks first such that children could properly inherit porperties.
# With the new API this seems to be in order, but I'm keeping this just in case for now. TODO: Could be used for optimization in the future.
tasks = sorted(tasks, key=lambda x: (
int(x.parent_id), x.order))
# If a type has changed, clean label for good measure
if label_id is not None:
# If a type has changed, clean all task labels for good measure
if next_action_label is not None:
if project_type_changed == 1 or section_type_changed == 1:
# Remove labels
[remove_label(item, label_id, overview_item_ids,
overview_item_labels) for item in items]
[remove_label(task, next_action_label, overview_task_ids,
overview_task_labels) for task in tasks]
# Remove parent types
for item in items:
item['parent_type'] = None
# for task in tasks:
# task.parent_type = None #TODO: METADATA
# For all items in this section
for item in items:
active_type = None # Reset
# For all tasks in this section
for task in tasks:
dominant_type = None # Reset
# Possible nottes routine for the future
# notes = api.notes.all() TODO: Quick notes test to see what the impact is?
# note_content = [x['content'] for x in notes if x['item_id'] == item['id']]
# print(note_content)
# Determine which child_items exist, both all and the ones that have not been checked yet
non_checked_items = list(
filter(lambda x: x['checked'] == 0, items))
child_items_all = list(
filter(lambda x: x['parent_id'] == item['id'], items))
child_items = list(
filter(lambda x: x['parent_id'] == item['id'], non_checked_items))
# Determine which child_tasks exist, both all and the ones that have not been checked yet
non_completed_tasks = list(
filter(lambda x: not x.is_completed, tasks))
child_tasks_all = list(
filter(lambda x: x.parent_id == task.id, tasks))
child_tasks = list(
filter(lambda x: x.parent_id == task.id, non_completed_tasks))
# Check if we need to (un)header entire item tree
header_all_in_i, unheader_all_in_i = check_header(item)
# Check if we need to (un)header entire task tree
header_all_in_t, unheader_all_in_t = check_header(task)
# Logic for applying and removing headers
if any([header_all_in_p, header_all_in_s, header_all_in_i]):
if item['content'][0] != '*':
item.update(content='* ' + item['content'])
for ci in child_items:
if not ci['content'].startswith('*'):
ci.update(content='* ' + ci['content'])
if any([unheader_all_in_p, unheader_all_in_s]):
if item['content'][0] == '*':
item.update(content=item['content'][2:])
if unheader_all_in_i:
[ci.update(content=ci['content'][2:])
for ci in child_items]
# Modify headers where needed
#TODO: DISABLED FOR NOW, FIX LATER
# modify_headers(header_all_in_p, unheader_all_in_p, header_all_in_s, unheader_all_in_s, header_all_in_t, unheader_all_in_t)
#TODO: Check is regeneration is still needed, now that it's part of core Todoist. Disabled for now.
# Logic for recurring lists
if not args.regeneration:
try:
# If old label is present, reset it
if item['r_tag'] == 1:
item['r_tag'] = 0
api.items.update(item['id'])
except:
pass
# if not args.regeneration:
# try:
# # If old label is present, reset it
# if item.r_tag == 1: #TODO: METADATA
# item.r_tag = 0 #TODO: METADATA
# api.items.update(item.id)
# except:
# pass
# If options turned on, start recurring lists logic
if args.regeneration is not None or args.end:
run_recurring_lists_logic(
args, api, item, child_items, child_items_all, regen_labels_id)
# # If options turned on, start recurring lists logic
# if args.regeneration is not None or args.end:
# run_recurring_lists_logic(
# args, api, item, child_items, child_items_all, regen_labels_id)
# If options turned on, start labelling logic
if label_id is not None:
# Skip processing an item if it has already been checked or is a header
if item['checked'] == 1:
if next_action_label is not None:
# Skip processing a task if it has already been checked or is a header
if task.is_completed:
continue
if item['content'].startswith('*'):
if task.content.startswith('*'):
# Remove next action label if it's still present
remove_label(item, label_id, overview_item_ids,overview_item_labels)
remove_label(task, next_action_label, overview_task_ids, overview_task_labels)
continue
# Check item type
item_type, item_type_changed = get_item_type(
args, item, project_type)
if item_type is not None:
# Check task type
task_type, task_type_changed = get_task_type(
args, task, project_type)
if task_type is not None:
logging.debug('Identified \'%s\' as %s type',
item['content'], item_type)
task.content, task_type)
# Determine hierarchy types for logic
hierarchy_types = [item_type,
hierarchy_types = [task_type,
section_type, project_type]
active_types = [type(x) != type(None)
hierarchy_boolean = [type(x) != type(None)
for x in hierarchy_types]
# If it is a parentless task
if item['parent_id'] == 0:
if active_types[0]:
# Do item types
active_type = item_type
if task.parent_id == 0:
if hierarchy_boolean[0]:
# Inherit task type
dominant_type = task_type
add_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
elif active_types[1]:
# Do section types
active_type = section_type
elif hierarchy_boolean[1]:
# Inherit section type
dominant_type = section_type
if section_type == 'sequential' or section_type == 's-p':
if not first_found_section:
add_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
first_found_section = True
elif section_type == 'parallel' or section_type == 'p-s':
add_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
elif active_types[2]:
# Do project types
active_type = project_type
elif hierarchy_boolean[2]:
# Inherit project type
dominant_type = project_type
if project_type == 'sequential' or project_type == 's-p':
if not first_found_project:
add_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
first_found_project = True
elif project_type == 'parallel' or project_type == 'p-s':
add_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
# Mark other conditions too
if first_found_section == False and active_types[1]:
if first_found_section == False and hierarchy_boolean[1]:
first_found_section = True
if first_found_project is False and active_types[2]:
if first_found_project is False and hierarchy_boolean[2]:
first_found_project = True
# If there are children
if len(child_items) > 0:
# Check if item state has changed, if so clean children for good measure
if item_type_changed == 1:
[remove_label(child_item, label_id, overview_item_ids, overview_item_labels)
for child_item in child_items]
if len(child_tasks) > 0:
# Check if task state has changed, if so clean children for good measure
if task_type_changed == 1:
[remove_label(child_task, next_action_label, overview_task_ids, overview_task_labels)
for child_task in child_tasks]
# If a sub-task, inherit parent task type
if item['parent_id'] !=0:
if task.parent_id !=0:
try:
active_type = item['parent_type']
dominant_type = task.parent_type #TODO: METADATA
except:
pass
# Process sequential tagged items (item_type can overrule project_type)
if active_type == 'sequential' or active_type == 'p-s':
for child_item in child_items:
# Process sequential tagged tasks (task_type can overrule project_type)
if dominant_type == 'sequential' or dominant_type == 'p-s':
for child_task in child_tasks:
# Ignore headered children
if child_item['content'].startswith('*'):
if child_task.content.startswith('*'):
continue
# Pass item_type down to the children
child_item['parent_type'] = active_type
# Pass task_type down to the children
child_task.parent_type = dominant_type
# Pass label down to the first child
if child_item['checked'] == 0 and label_id in item['labels']:
if not child_task.is_completed and next_action_label in task.labels:
add_label(
child_item, label_id, overview_item_ids, overview_item_labels)
child_task, next_action_label, overview_task_ids, overview_task_labels)
remove_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
else:
# Clean for good measure
remove_label(
child_item, label_id, overview_item_ids, overview_item_labels)
child_task, next_action_label, overview_task_ids, overview_task_labels)
# Process parallel tagged items or untagged parents
elif active_type == 'parallel' or (active_type == 's-p' and label_id in item['labels']):
# Process parallel tagged tasks or untagged parents
elif dominant_type == 'parallel' or (dominant_type == 's-p' and next_action_label in task.labels):
remove_label(
item, label_id, overview_item_ids, overview_item_labels)
for child_item in child_items:
task, next_action_label, overview_task_ids, overview_task_labels)
for child_task in child_tasks:
# Ignore headered children
if child_item['content'].startswith('*'):
if child_task.content.startswith('*'):
continue
child_item['parent_type'] = active_type
if child_item['checked'] == 0:
# child_first_found = True
child_task.parent_type = dominant_type #TODO: METADATA
if not child_task.is_completed:
add_label(
child_item, label_id, overview_item_ids, overview_item_labels)
child_task, next_action_label, overview_task_ids, overview_task_labels)
# Remove labels based on start / due dates
# If item is too far in the future, remove the next_action tag and skip
# If task is too far in the future, remove the next_action tag and skip #TODO: FIX THIS
try:
if args.hide_future > 0 and 'due' in item.data and item['due'] is not None:
if args.hide_future > 0 and 'due' in task.data and task.due is not None:
due_date = datetime.strptime(
item['due']['date'][:10], "%Y-%m-%d")
task.due['date'][:10], "%Y-%m-%d")
future_diff = (
due_date - datetime.today()).days
if future_diff >= args.hide_future:
remove_label(
item, label_id, overview_item_ids, overview_item_labels)
task, next_action_label, overview_task_ids, overview_task_labels)
continue
except:
# Hide-future not set, skip
@ -836,15 +850,15 @@ def autodoist_magic(args, api, label_id, regen_labels_id):
# If start-date has not passed yet, remove label
try:
f1 = item['content'].find('start=')
f2 = item['content'].find('start=due-')
f1 = task.content.find('start=')
f2 = task.content.find('start=due-')
if f1 > -1 and f2 == -1:
f_end = item['content'][f1+6:].find(' ')
f_end = task.content[f1+6:].find(' ')
if f_end > -1:
start_date = item['content'][f1 +
start_date = task.content[f1 +
6:f1+6+f_end]
else:
start_date = item['content'][f1+6:]
start_date = task.content[f1+6:]
# If start-date hasen't passed, remove all labels
start_date = datetime.strptime(
@ -853,39 +867,39 @@ def autodoist_magic(args, api, label_id, regen_labels_id):
datetime.today()-start_date).days
if future_diff < 0:
remove_label(
item, label_id, overview_item_ids, overview_item_labels)
[remove_label(child_item, label_id, overview_item_ids,
overview_item_labels) for child_item in child_items]
task, next_action_label, overview_task_ids, overview_task_labels)
[remove_label(child_task, next_action_label, overview_task_ids,
overview_task_labels) for child_task in child_tasks]
continue
except:
logging.warning(
'Wrong start-date format for item: "%s". Please use "start=<DD-MM-YYYY>"', item['content'])
'Wrong start-date format for task: "%s". Please use "start=<DD-MM-YYYY>"', task.content)
continue
# Recurring task friendly - remove label with relative change from due date
# Recurring task friendly - remove label with relative change from due date #TODO Fix this logic
try:
f = item['content'].find('start=due-')
f = task.content.find('start=due-')
if f > -1:
f1a = item['content'].find(
f1a = task.content.find(
'd') # Find 'd' from 'due'
f1b = item['content'].rfind(
f1b = task.content.rfind(
'd') # Find 'd' from days
f2 = item['content'].find('w')
f_end = item['content'][f+10:].find(' ')
f2 = task.content.find('w')
f_end = task.content[f+10:].find(' ')
if f_end > -1:
offset = item['content'][f+10:f+10+f_end-1]
offset = task.content[f+10:f+10+f_end-1]
else:
offset = item['content'][f+10:-1]
offset = task.content[f+10:-1]
try:
item_due_date = item['due']['date'][:10]
item_due_date = datetime.strptime(
item_due_date, '%Y-%m-%d')
task_due_date = task.due['date'][:10]
task_due_date = datetime.strptime(
task_due_date, '%Y-%m-%d')
except:
logging.warning(
'No due date to determine start date for item: "%s".', item['content'])
'No due date to determine start date for task: "%s".', task.content)
continue
if f1a != f1b and f1b > -1: # To make sure it doesn't trigger if 'w' is chosen
@ -894,22 +908,35 @@ def autodoist_magic(args, api, label_id, regen_labels_id):
td = timedelta(weeks=int(offset))
# If we're not in the offset from the due date yet, remove all labels
start_date = item_due_date - td
start_date = task_due_date - td
future_diff = (
datetime.today()-start_date).days
if future_diff < 0:
remove_label(
item, label_id, overview_item_ids, overview_item_labels)
[remove_label(child_item, label_id, overview_item_ids,
overview_item_labels) for child_item in child_items]
task, next_action_label, overview_task_ids, overview_task_labels)
[remove_label(child_task, next_action_label, overview_task_ids,
overview_task_labels) for child_task in child_tasks]
continue
except:
logging.warning(
'Wrong start-date format for item: %s. Please use "start=due-<NUM><d or w>"', item['content'])
'Wrong start-date format for task: %s. Please use "start=due-<NUM><d or w>"', task.content)
continue
return overview_item_ids, overview_item_labels
return overview_task_ids, overview_task_labels
# Connect to SQLite database
def create_connection(path):
connection = None
try:
connection = sqlite3.connect(path)
print("Connection to SQLite DB successful")
except Error as e:
print(f"The error '{e}' occurred")
return connection
# Main
@ -933,13 +960,13 @@ def main():
parser.add_argument(
'-d', '--delay', help='specify the delay in seconds between syncs (default 5).', default=5, type=int)
parser.add_argument(
'-pp', '--pp_suffix', help='change suffix for parallel-parallel labeling (default "//").', default='//')
'-pp', '--pp_suffix', help='change suffix for parallel-parallel labeling (default "==").', default='==')
parser.add_argument(
'-ss', '--ss_suffix', help='change suffix for sequential-sequential labeling (default "--").', default='--')
parser.add_argument(
'-ps', '--ps_suffix', help='change suffix for parallel-sequential labeling (default "/-").', default='/-')
'-ps', '--ps_suffix', help='change suffix for parallel-sequential labeling (default "=-").', default='=-')
parser.add_argument(
'-sp', '--sp_suffix', help='change suffix for sequential-parallel labeling (default "-/").', default='-/')
'-sp', '--sp_suffix', help='change suffix for sequential-parallel labeling (default "-=").', default='-=')
parser.add_argument(
'-df', '--dateformat', help='strptime() format of starting date (default "%%d-%%m-%%Y").', default='%d-%m-%Y')
parser.add_argument(
@ -955,6 +982,9 @@ def main():
args = parser.parse_args()
# #TODO: Temporary disable this feature for now. Find a way to see completed tasks first, since REST API v2 lost this funcionality.
args.regeneration = 0
# Addition of regeneration labels
args.regen_label_names = ('Regen_off', 'Regen_all',
'Regen_all_if_completed')
@ -977,25 +1007,25 @@ def main():
check_for_update(current_version)
# Initialise api
api, label_id, regen_labels_id = initialise(args)
api = initialise(args)
# Start main loop
while True:
start_time = time.time()
# sync(api)
# Evaluate projects, sections, and items
overview_item_ids, overview_item_labels = autodoist_magic(
args, api, label_id, regen_labels_id)
# Evaluate projects, sections, and tasks
overview_task_ids, overview_task_labels = autodoist_magic(
args, api, args.label, args.regen_label_names)
# Commit the queue with changes
if label_id is not None:
update_labels(api, label_id, overview_item_ids,
overview_item_labels)
# Commit next action label changes
if args.label is not None:
updated_ids = update_labels(api, overview_task_ids,
overview_task_labels)
if len(updated_ids):
len_api_q = len(updated_ids)
if len(api.queue):
len_api_q = len(api.queue)
api.commit()
if len_api_q == 1:
logging.info(
'%d change committed to Todoist.', len_api_q)

4
requirements.txt
View File

@ -1,2 +1,2 @@
requests>=2.25.1
todoist_python>=8.1.3
requests>=2.28.1
todoist_api_python>=2.0.2