2024-08-23 23:47:24 +02:00
|
|
|
"""
|
|
|
|
Inkycal Todoist Module
|
|
|
|
Copyright by aceinnolab
|
|
|
|
"""
|
|
|
|
import arrow
|
2024-08-24 11:26:52 +02:00
|
|
|
import json
|
|
|
|
import logging
|
|
|
|
import requests
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
from inkycal.modules.template import inkycal_module
|
|
|
|
from inkycal.custom import *
|
|
|
|
|
|
|
|
from todoist_api_python.api import TodoistAPI
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
2024-08-24 11:26:52 +02:00
|
|
|
class LoginVikunja():
|
|
|
|
def __init__(self, username, password, totp_passcode=None, token=None, api_url='http://192.168.50.10:3456/api/v1/'):
|
|
|
|
self.username = username
|
|
|
|
self.password = password
|
|
|
|
self.totp_passcode = totp_passcode
|
|
|
|
self.token = None
|
|
|
|
self.api_url = api_url
|
|
|
|
self._access_token = token
|
|
|
|
if self._access_token is None:
|
|
|
|
self._access_token = self.get_token()
|
|
|
|
|
|
|
|
def _create_url(self, path):
|
|
|
|
return self.api_url + path
|
|
|
|
|
|
|
|
"""returns the token from the login request"""
|
|
|
|
def _post_login_request(self, username, password, totp_passcode):
|
|
|
|
login_url = self._create_url('login')
|
|
|
|
payload = {
|
|
|
|
'long_token': True,
|
|
|
|
'username': username,
|
|
|
|
'password': password,
|
|
|
|
'totp_passcode': totp_passcode
|
|
|
|
}
|
|
|
|
return requests.post(login_url, json=payload, timeout=5)
|
|
|
|
|
|
|
|
def _get_access_token(self):
|
|
|
|
if not self._access_token:
|
|
|
|
token_json = self._post_login_request(self.username, self.password, self.totp_passcode)
|
|
|
|
if token_json.status_code == 200:
|
|
|
|
token = json.loads(token_json.text)
|
|
|
|
self._access_token = token['token']
|
|
|
|
else:
|
|
|
|
raise Exception('Login failed')
|
|
|
|
return self._access_token
|
|
|
|
|
|
|
|
def get_token(self):
|
|
|
|
return self._get_access_token()
|
|
|
|
|
|
|
|
def get_headers(self):
|
|
|
|
return {'Authorization': 'Bearer ' + self._get_access_token()}
|
|
|
|
|
|
|
|
class ApiVikunja():
|
|
|
|
def __init__(self, username, password, totp_passcode=None, token=None, api_url='http://192.168.50.10:3456/api/v1/'):
|
|
|
|
self.username = username
|
|
|
|
self.password = password
|
|
|
|
self.totp_passcode = totp_passcode
|
|
|
|
self.token = None
|
|
|
|
self.api_url = api_url
|
|
|
|
self._cache = {'projects': None, 'tasks': None, 'labels': None}
|
|
|
|
self._login = LoginVikunja(username, password, totp_passcode, token, api_url)
|
|
|
|
def _create_url(self, path):
|
|
|
|
return self.api_url + path
|
|
|
|
|
|
|
|
def _to_json(self, response):
|
|
|
|
try:
|
|
|
|
return response.json()
|
|
|
|
except Exception as e:
|
|
|
|
logger.error(f'Error parsing json: {e}')
|
|
|
|
raise e
|
|
|
|
|
|
|
|
def _get_json(self, url, params=None, headers=None):
|
|
|
|
if params is None:
|
|
|
|
params = {}
|
|
|
|
response = requests.get(url, params=params, headers=headers, timeout=5)
|
|
|
|
response.raise_for_status()
|
|
|
|
json_result = self._to_json(response)
|
|
|
|
total_pages = int(response.headers.get('x-pagination-total-pages', 1))
|
|
|
|
if total_pages > 1:
|
|
|
|
logger.debug('Trying to get all pages')
|
|
|
|
for page in range(2, total_pages + 1):
|
|
|
|
logger.debug(f'Getting page {page}')
|
|
|
|
params.update({'page': page})
|
|
|
|
response = requests.get(url, params=params, headers=headers, timeout=5)
|
|
|
|
response.raise_for_status()
|
|
|
|
json_result = json_result + self._to_json(response)
|
|
|
|
return json_result
|
|
|
|
|
|
|
|
def get_projects(self):
|
2024-08-26 11:30:20 +02:00
|
|
|
# if self._cache['projects'] is None:
|
|
|
|
self._cache['projects'] = self._get_json(self._create_url('projects'), headers=self._login.get_headers())
|
2024-08-24 11:26:52 +02:00
|
|
|
return self._cache['projects']
|
2024-08-24 15:03:43 +02:00
|
|
|
|
2024-08-24 11:26:52 +02:00
|
|
|
def get_tasks(self, exclude_completed=True):
|
2024-08-26 11:30:20 +02:00
|
|
|
# if self._cache['tasks'] is None:
|
|
|
|
url = self._create_url('tasks/all')
|
|
|
|
params = {'filter': 'done=false'} if exclude_completed else {}
|
|
|
|
self._cache['tasks'] = self._get_json(url, params, headers=self._login.get_headers()) or []
|
2024-08-24 11:26:52 +02:00
|
|
|
return self._cache['tasks']
|
|
|
|
|
|
|
|
|
2024-08-23 23:47:24 +02:00
|
|
|
|
2024-08-24 15:03:43 +02:00
|
|
|
class Vikunja(inkycal_module):
|
2024-08-23 23:47:24 +02:00
|
|
|
"""Todoist api class
|
|
|
|
parses todos from the todoist api.
|
|
|
|
"""
|
|
|
|
|
|
|
|
name = "Vikunja API - show your todos from Vikunja"
|
|
|
|
|
|
|
|
requires = {
|
|
|
|
'url-frontend': {
|
|
|
|
"label": "Please enter your Vikunja URL",
|
|
|
|
},
|
|
|
|
'url-backend': {
|
|
|
|
"label": "Please enter your Vikunja URL",
|
|
|
|
},
|
2024-08-24 15:03:43 +02:00
|
|
|
'username': {
|
|
|
|
"label": "Please enter your Vikunja username",
|
|
|
|
},
|
|
|
|
'password': {
|
|
|
|
"label": "Please enter your Vikunja password",
|
|
|
|
},
|
2024-08-23 23:47:24 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
optional = {
|
|
|
|
'project_filter': {
|
|
|
|
"label": "Show Todos only from following project (separated by a comma). Leave empty to show " +
|
|
|
|
"todos from all projects",
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
def __init__(self, config):
|
|
|
|
"""Initialize inkycal_rss module"""
|
|
|
|
|
|
|
|
super().__init__(config)
|
|
|
|
|
|
|
|
config = config['config']
|
|
|
|
|
|
|
|
# Check if all required parameters are present
|
|
|
|
for param in self.requires:
|
|
|
|
if param not in config:
|
|
|
|
raise Exception(f'config is missing {param}')
|
|
|
|
|
|
|
|
# module specific parameters
|
|
|
|
self.frontend_url = config['url-frontend']
|
|
|
|
self.backend_url = config['url-backend']
|
|
|
|
|
|
|
|
# if project filter is set, initialize it
|
|
|
|
if config['project_filter'] and isinstance(config['project_filter'], str):
|
|
|
|
self.project_filter = config['project_filter'].split(',')
|
|
|
|
else:
|
|
|
|
self.project_filter = config['project_filter']
|
|
|
|
|
2024-08-24 15:03:43 +02:00
|
|
|
# self._api = TodoistAPI(config['api_key'])
|
|
|
|
self._vikunja_api = ApiVikunja(config['username'], config['password'], None, None, config['url-backend'])
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
# give an OK message
|
|
|
|
logger.debug(f'{__name__} loaded')
|
|
|
|
|
|
|
|
def _validate(self):
|
|
|
|
"""Validate module-specific parameters"""
|
|
|
|
if not isinstance(self.api_key, str):
|
|
|
|
print('api_key has to be a string: "Yourtopsecretkey123" ')
|
2024-08-24 11:26:52 +02:00
|
|
|
|
|
|
|
def get_projects():
|
|
|
|
|
|
|
|
pass
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
def generate_image(self):
|
|
|
|
"""Generate image for this module"""
|
|
|
|
|
|
|
|
# Define new image size with respect to padding
|
|
|
|
im_width = int(self.width - (2 * self.padding_left))
|
|
|
|
im_height = int(self.height - (2 * self.padding_top))
|
|
|
|
im_size = im_width, im_height
|
|
|
|
logger.debug(f'Image size: {im_size}')
|
|
|
|
|
|
|
|
# Create an image for black pixels and one for coloured pixels
|
|
|
|
im_black = Image.new('RGB', size=im_size, color='white')
|
|
|
|
im_colour = Image.new('RGB', size=im_size, color='white')
|
|
|
|
|
|
|
|
# Check if internet is available
|
|
|
|
if internet_available():
|
|
|
|
logger.info('Connection test passed')
|
|
|
|
else:
|
|
|
|
logger.error("Network not reachable. Please check your connection.")
|
|
|
|
raise NetworkNotReachableError
|
|
|
|
|
|
|
|
# Set some parameters for formatting todos
|
|
|
|
line_spacing = 1
|
|
|
|
text_bbox_height = self.font.getbbox("hg")
|
|
|
|
line_height = text_bbox_height[3] + line_spacing
|
|
|
|
line_width = im_width
|
|
|
|
max_lines = im_height // line_height
|
|
|
|
|
|
|
|
# Calculate padding from top so the lines look centralised
|
|
|
|
spacing_top = int(im_height % line_height / 2)
|
|
|
|
|
|
|
|
# Calculate line_positions
|
|
|
|
line_positions = [
|
|
|
|
(0, spacing_top + _ * line_height) for _ in range(max_lines)]
|
|
|
|
|
|
|
|
# Get all projects by name and id
|
2024-08-24 15:03:43 +02:00
|
|
|
# all_projects = self._api.get_projects()
|
|
|
|
# filtered_project_ids_and_names = {project.id: project.name for project in all_projects}
|
|
|
|
# all_active_tasks = self._api.get_tasks()
|
|
|
|
all_projects = self._vikunja_api.get_projects()
|
|
|
|
all_active_tasks = self._vikunja_api.get_tasks()
|
|
|
|
all_active_tasks = [task for task in all_active_tasks if task['done'] == False]
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
logger.debug(f"all_projects: {all_projects}")
|
2024-08-24 15:03:43 +02:00
|
|
|
logger.debug(f"all_active_tasks: {all_active_tasks}")
|
|
|
|
print(f"all_projects: {all_projects}")
|
|
|
|
print(f"all_active_tasks: {all_active_tasks}")
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
# Filter entries in all_projects if filter was given
|
|
|
|
if self.project_filter:
|
2024-08-24 15:03:43 +02:00
|
|
|
# filtered_projects = [project for project in all_projects if project.name in self.project_filter]
|
|
|
|
filtered_projects = [project for project in all_projects if project['title'] in self.project_filter]
|
|
|
|
filtered_project_ids_and_names = {project['id']: project['title'] for project in filtered_projects}
|
2024-08-23 23:47:24 +02:00
|
|
|
filtered_project_ids = [project for project in filtered_project_ids_and_names]
|
|
|
|
logger.debug(f"filtered projects: {filtered_projects}")
|
2024-08-24 15:03:43 +02:00
|
|
|
print(f"filtered projects: {filtered_projects}")
|
|
|
|
print(f"filtered_project_ids_and_names: {filtered_project_ids_and_names}")
|
|
|
|
print(f"filtered_project_ids: {filtered_project_ids}")
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
# If filter was activated and no project was found with that name,
|
|
|
|
# raise an exception to avoid showing a blank image
|
|
|
|
if not filtered_projects:
|
|
|
|
logger.error('No project found from project filter!')
|
|
|
|
logger.error('Please double check spellings in project_filter')
|
|
|
|
raise Exception('No matching project found in filter. Please '
|
|
|
|
'double check spellings in project_filter or leave'
|
|
|
|
'empty')
|
|
|
|
# filtered version of all active tasks
|
2024-08-24 15:03:43 +02:00
|
|
|
all_active_tasks = [task for task in all_active_tasks if task['project_id'] in filtered_project_ids]
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
# Simplify the tasks for faster processing
|
|
|
|
simplified = [
|
|
|
|
{
|
2024-08-24 15:03:43 +02:00
|
|
|
'name': task['title'],
|
|
|
|
'due': arrow.get(task['due_date']).format("D-MMM-YY") if 'due_date' in task and task['due_date'][:2] != '00' else "",
|
|
|
|
'priority': task['priority'],
|
|
|
|
'project': filtered_project_ids_and_names[task['project_id']]
|
2024-08-23 23:47:24 +02:00
|
|
|
}
|
|
|
|
for task in all_active_tasks
|
|
|
|
]
|
|
|
|
|
|
|
|
logger.debug(f'simplified: {simplified}')
|
2024-08-24 15:03:43 +02:00
|
|
|
print(f'simplified: {simplified}')
|
2024-08-23 23:47:24 +02:00
|
|
|
|
|
|
|
project_lengths = []
|
|
|
|
due_lengths = []
|
|
|
|
|
|
|
|
for task in simplified:
|
|
|
|
if task["project"]:
|
|
|
|
project_lengths.append(int(self.font.getlength(task['project']) * 1.1))
|
|
|
|
if task["due"]:
|
|
|
|
due_lengths.append(int(self.font.getlength(task['due']) * 1.1))
|
|
|
|
|
|
|
|
# Get maximum width of project names for selected font
|
|
|
|
project_offset = int(max(project_lengths)) if project_lengths else 0
|
|
|
|
|
|
|
|
# Get maximum width of project dues for selected font
|
|
|
|
due_offset = int(max(due_lengths)) if due_lengths else 0
|
|
|
|
|
|
|
|
# create a dict with names of filtered groups
|
|
|
|
groups = {group_name:[] for group_name in filtered_project_ids_and_names.values()}
|
|
|
|
for task in simplified:
|
|
|
|
group_of_current_task = task["project"]
|
|
|
|
if group_of_current_task in groups:
|
|
|
|
groups[group_of_current_task].append(task)
|
|
|
|
|
|
|
|
logger.debug(f"grouped: {groups}")
|
|
|
|
|
|
|
|
# Add the parsed todos on the image
|
|
|
|
cursor = 0
|
|
|
|
for name, todos in groups.items():
|
|
|
|
if todos:
|
|
|
|
for todo in todos:
|
|
|
|
if cursor < max_lines:
|
|
|
|
line_x, line_y = line_positions[cursor]
|
|
|
|
|
|
|
|
if todo['project']:
|
|
|
|
# Add todos project name
|
|
|
|
write(
|
|
|
|
im_colour, line_positions[cursor],
|
|
|
|
(project_offset, line_height),
|
|
|
|
todo['project'], font=self.font, alignment='left')
|
|
|
|
|
|
|
|
# Add todos due if not empty
|
|
|
|
if todo['due']:
|
|
|
|
write(
|
|
|
|
im_black,
|
|
|
|
(line_x + project_offset, line_y),
|
|
|
|
(due_offset, line_height),
|
|
|
|
todo['due'], font=self.font, alignment='left')
|
|
|
|
|
|
|
|
if todo['name']:
|
|
|
|
# Add todos name
|
|
|
|
write(
|
|
|
|
im_black,
|
|
|
|
(line_x + project_offset + due_offset, line_y),
|
|
|
|
(im_width - project_offset - due_offset, line_height),
|
|
|
|
todo['name'], font=self.font, alignment='left')
|
|
|
|
|
|
|
|
cursor += 1
|
|
|
|
else:
|
|
|
|
logger.error('More todos than available lines')
|
|
|
|
break
|
|
|
|
|
|
|
|
# return the images ready for the display
|
|
|
|
return im_black, im_colour
|