things3.things3.Things3.get_mit() - Code Metrics - Inspection of "Merge pull request #10 from AlexanderWillner/move-..." - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( dd0bda...e4c11f )

by Alexander

created 2020-04-22 08:41 UTC

things3.things3.Things3.get_mit() A

↳ Parent: things3.things3

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	3
rs	10
c	0
b	0
f	0

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""Simple read-only API for Things 3."""

from __future__ import print_function

__author__ = "Alexander Willner"
__copyright__ = "2020 Alexander Willner"
__credits__ = ["Alexander Willner"]
__license__ = "Apache License 2.0"
__version__ = "2.5.0.dev0"
__maintainer__ = "Alexander Willner"
__email__ = "[email protected]"
__status__ = "Development"

import sqlite3
import sys
from random import shuffle
from os import environ
import getpass


# pylint: disable=R0904
class Things3():
    """Simple read-only API for Things 3."""
    # Variables
    debug = False
    database = None
    json = False
    filter = ""
    tag_waiting = "Waiting" if not environ.get('TAG_WAITING') \
        else environ.get('TAG_WAITING')
    tag_mit = "MIT" if not environ.get('TAG_MIT') \
        else environ.get('TAG_MIT')
    tag_cleanup = "Cleanup" if not environ.get('TAG_CLEANUP') \
        else environ.get('TAG_CLEANUP')
    anonymize = bool(environ.get('ANONYMIZE'))

    # Database info
    FILE_DB = '/Library/Containers/'\
              'com.culturedcode.ThingsMac/Data/Library/'\
              'Application Support/Cultured Code/Things/Things.sqlite3'
    FILE_SQLITE = '/Users/' + getpass.getuser() + FILE_DB \
        if not environ.get('THINGSDB') else environ.get('THINGSDB')

    TABLE_TASK = "TMTask"
    TABLE_AREA = "TMArea"
    TABLE_TAG = "TMTag"
    TABLE_TASKTAG = "TMTaskTag"
    DATE_CREATE = "creationDate"
    DATE_MOD = "userModificationDate"
    DATE_DUE = "dueDate"
    DATE_START = "startDate"
    DATE_STOP = "stopDate"
    IS_INBOX = "start = 0"
    IS_ANYTIME = "start = 1"
    IS_SOMEDAY = "start = 2"
    IS_SCHEDULED = f"{DATE_START} IS NOT NULL"
    IS_NOT_SCHEDULED = f"{DATE_START} IS NULL"
    IS_DUE = f"{DATE_DUE} IS NOT NULL"
    IS_NOT_DUE = f"{DATE_DUE} IS NULL"
    IS_RECURRING = "recurrenceRule IS NOT NULL"
    IS_NOT_RECURRING = "recurrenceRule IS NULL"
    IS_TASK = "type = 0"
    IS_PROJECT = "type = 1"
    IS_HEADING = "type = 2"
    IS_TRASHED = "trashed = 1"
    IS_NOT_TRASHED = "trashed = 0"
    IS_OPEN = "status = 0"
    IS_CANCELLED = "status = 2"
    IS_DONE = "status = 3"
    MODE_TASK = "type = 0"
    MODE_PROJECT = "type = 1"

    def __init__(self,
                 database=FILE_SQLITE,
                 tag_waiting='Waiting',
                 tag_mit='MIT',
                 json=False):
        self.database = database if database is not None else self.FILE_SQLITE
        self.tag_mit = tag_mit
        self.tag_waiting = tag_waiting
        self.json = json

    @staticmethod
    def anonymize_string(string):
        """Scramble text."""
        if string is None:
            return None
        string = list(string)
        shuffle(string)
        string = ''.join(string)
        return string

    @staticmethod
    def dict_factory(cursor, row):
        """Convert SQL result into a dictionary"""
        dictionary = {}
        for idx, col in enumerate(cursor.description):
            dictionary[col[0]] = row[idx]
        return dictionary

    def anonymize_tasks(self, tasks):
        """Scramble output for screenshots."""
        if self.anonymize:
            for task in tasks:
                task['title'] = self.anonymize_string(task['title'])
                task['context'] = self.anonymize_string(task['context'])
        return tasks

    def get_inbox(self):
        """Get all tasks from the inbox."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_INBOX}
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_today(self):
        """Get all tasks from the today list."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                (TASK.{self.IS_ANYTIME} OR (
                     TASK.{self.IS_SOMEDAY} AND
                     TASK.{self.DATE_START} <= strftime('%s', 'now')
                     )
                ) AND
                TASK.{self.IS_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_someday(self):
        """Get someday tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_SOMEDAY} AND
                TASK.{self.IS_NOT_SCHEDULED} AND
                TASK.{self.IS_NOT_RECURRING} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC, TASK.creationdate DESC
                """
        return self.get_rows(query)

    def get_upcoming(self):
        """Get upcoming tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_SOMEDAY} AND
                TASK.{self.IS_SCHEDULED} AND
                TASK.{self.IS_NOT_RECURRING} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.startdate, TASK.todayIndex
                """
        return self.get_rows(query)

    def get_waiting(self):
        """Get waiting tasks."""
        return self.get_tag(self.tag_waiting)

    def get_mit(self):
        """Get most important tasks."""
        return self.get_tag(self.tag_mit)

    def get_tag(self, tag):
        """Get task with specific tag"""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_NOT_RECURRING} AND
                TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
                             WHERE title='{tag}'
                          )
                AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_anytime(self):
        """Get anytime tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_ANYTIME} AND
                TASK.{self.IS_NOT_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_ANYTIME} AND
                            PROJECT.{self.IS_NOT_SCHEDULED} AND
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_ANYTIME} AND
                            HEADPROJ.{self.IS_NOT_SCHEDULED} AND
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        if self.filter:
            # ugly hack for Kanban task view on project
            query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_ANYTIME} AND
                TASK.{self.IS_NOT_SCHEDULED} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.duedate DESC , TASK.todayIndex
                """
        return self.get_rows(query)

    def get_completed(self):
        """Get completed tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_DONE}
                ORDER BY TASK.{self.DATE_STOP}
                """
        return self.get_rows(query)

    def get_cancelled(self):
        """Get cancelled tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_CANCELLED}
                ORDER BY TASK.{self.DATE_STOP}
                """
        return self.get_rows(query)

    def get_trashed(self):
        """Get trashed tasks."""
        query = f"""
                TASK.{self.IS_TRASHED} AND
                TASK.{self.IS_TASK}
                ORDER BY TASK.{self.DATE_STOP}
                """
        return self.get_rows(query)

    def get_projects(self):
        """Get projects."""
        query = f"""
                SELECT
                    TASK.uuid,
                    TASK.title,
                    NULL as context
                FROM
                    {self.TABLE_TASK} AS TASK
                WHERE
                    TASK.{self.IS_NOT_TRASHED} AND
                    TASK.{self.IS_PROJECT} AND
                    TASK.{self.IS_OPEN}
                ORDER BY TASK.title
                """
        return self.execute_query(query)

    def get_areas(self):
        """Get areas."""
        query = f"""
            SELECT
                AREA.uuid AS uuid,
                AREA.title AS title
            FROM
                {self.TABLE_AREA} AS AREA
            ORDER BY AREA.title
            """
        return self.execute_query(query)

    def get_all(self):
        """Get all tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                """
        return self.get_rows(query)

    def get_due(self):
        """Get due tasks."""
        query = f"""
                TASK.{self.IS_NOT_TRASHED} AND
                TASK.{self.IS_TASK} AND
                TASK.{self.IS_OPEN} AND
                TASK.{self.IS_DUE} AND (
                    (
                        PROJECT.title IS NULL OR (
                            PROJECT.{self.IS_NOT_TRASHED}
                        )
                    ) AND (
                        HEADPROJ.title IS NULL OR (
                            HEADPROJ.{self.IS_NOT_TRASHED}
                        )
                    )
                )
                ORDER BY TASK.{self.DATE_DUE}
                """
        return self.get_rows(query)

    def get_lint(self):
        """Get tasks that float around"""
        query = f"""
            TASK.{self.IS_NOT_TRASHED} AND
            TASK.{self.IS_OPEN} AND
            TASK.{self.IS_TASK} AND
            (TASK.{self.IS_SOMEDAY} OR TASK.{self.IS_ANYTIME}) AND
            TASK.project IS NULL AND
            TASK.area IS NULL AND
            TASK.actionGroup IS NULL
            """
        return self.get_rows(query)

    def get_empty_projects(self):
        """Get projects that are empty"""
        query = f"""
            TASK.{self.IS_NOT_TRASHED} AND
            TASK.{self.IS_OPEN} AND
            TASK.{self.IS_PROJECT} AND
            (TASK.{self.IS_ANYTIME} OR TASK.{self.IS_SCHEDULED})
            GROUP BY TASK.uuid
            HAVING
                (SELECT COUNT(uuid)
                 FROM TMTask AS PROJECT_TASK
                 WHERE
                   PROJECT_TASK.project = TASK.uuid AND
                   PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                   PROJECT_TASK.{self.IS_OPEN} AND
                   PROJECT_TASK.{self.IS_ANYTIME}
                ) = 0
            """
        return self.get_rows(query)

    def get_largest_projects(self):
        """Get projects that are empty"""
        query = f"""
            SELECT
                TASK.uuid,
                TASK.title AS title,
                creationDate AS created,
                userModificationDate AS modified,
                (SELECT COUNT(uuid)
                 FROM TMTask AS PROJECT_TASK
                 WHERE
                   PROJECT_TASK.project = TASK.uuid AND
                   PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                   PROJECT_TASK.{self.IS_OPEN}
                ) AS tasks
            FROM
                {self.TABLE_TASK} AS TASK
            WHERE
               TASK.{self.IS_NOT_TRASHED} AND
               TASK.{self.IS_OPEN} AND
               TASK.{self.IS_PROJECT}
            GROUP BY TASK.uuid
            ORDER BY tasks DESC
            """
        return self.execute_query(query)

    def get_daystats(self):
        """Get a history of task activities"""
        days = 365
        query = f"""
                WITH RECURSIVE timeseries(x) AS (
                    SELECT 0
                    UNION ALL
                    SELECT x+1 FROM timeseries
                    LIMIT {days}
                )
                SELECT
                    date(julianday("now", "-{days} days"),
                         "+" || x || " days") as date,
                    CREATED.TasksCreated as created,
                    CLOSED.TasksClosed as completed,
                    CANCELLED.TasksCancelled as cancelled,
                    TRASHED.TasksTrashed as trashed
                FROM timeseries
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksCreated,
                        date(creationDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                    AS CREATED ON CREATED.DAY = date
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksCancelled,
                        date(stopDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_CANCELLED} AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                        AS CANCELLED ON CANCELLED.DAY = date
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksTrashed,
                        date(userModificationDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_TRASHED} AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                        AS TRASHED ON TRASHED.DAY = date
                LEFT JOIN
                    (SELECT COUNT(uuid) AS TasksClosed,
                        date(stopDate,"unixepoch") AS DAY
                        FROM {self.TABLE_TASK} AS TASK
                        WHERE DAY NOT NULL
                          AND TASK.{self.IS_DONE} AND TASK.{self.IS_TASK}
                        GROUP BY DAY)
                        AS CLOSED ON CLOSED.DAY = date
                """
        return self.execute_query(query)

    def get_minutes_today(self):
        """Count the planned minutes for today."""
        query = f"""
                SELECT
                    SUM(TAG.title) AS minutes
                FROM
                    {self.TABLE_TASK} AS TASK
                LEFT OUTER JOIN
                TMTask PROJECT ON TASK.project = PROJECT.uuid
                LEFT OUTER JOIN
                    TMArea AREA ON TASK.area = AREA.uuid
                LEFT OUTER JOIN
                    TMTask HEADING ON TASK.actionGroup = HEADING.uuid
                LEFT OUTER JOIN
                    TMTask HEADPROJ ON HEADING.project = HEADPROJ.uuid
                LEFT OUTER JOIN
                    TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
                LEFT OUTER JOIN
                    TMTag TAG ON TAGS.tags = TAG.uuid
                WHERE
                    printf("%d", TAG.title) = TAG.title AND
                    TASK.{self.IS_NOT_TRASHED} AND
                    TASK.{self.IS_TASK} AND
                    TASK.{self.IS_OPEN} AND
                    TASK.{self.IS_ANYTIME} AND
                    TASK.{self.IS_SCHEDULED} AND (
                        (
                            PROJECT.title IS NULL OR (
                                PROJECT.{self.IS_NOT_TRASHED}
                            )
                        ) AND (
                            HEADPROJ.title IS NULL OR (
                                HEADPROJ.{self.IS_NOT_TRASHED}
                            )
                        )
                    )
                """
        return self.execute_query(query)

    def get_cleanup(self):
        """Tasks and projects that need work."""
        result = []
        result.extend(self.get_lint())
        result.extend(self.get_empty_projects())
        result.extend(self.get_tag(self.tag_cleanup))
        return result

    @staticmethod
    def get_not_implemented():
        """Not implemented warning."""
        return [{"title": "not implemented"}]

    def get_rows(self, sql):
        """Query Things database."""

        sql = f"""
            SELECT DISTINCT
                TASK.uuid,
                TASK.title,
                CASE
                    WHEN AREA.title IS NOT NULL THEN AREA.title
                    WHEN PROJECT.title IS NOT NULL THEN PROJECT.title
                    WHEN HEADING.title IS NOT NULL THEN HEADING.title
                END AS context,
                CASE
                    WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
                    WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
                END AS context_uuid,
                CASE
                    WHEN TASK.recurrenceRule IS NULL
                    THEN date(TASK.dueDate,"unixepoch")
                ELSE NULL
                END AS due,
                date(TASK.creationDate,"unixepoch") as created,
                date(TASK.userModificationDate,"unixepoch") as modified,
                date(TASK.startDate,"unixepoch") as started,
                date(TASK.stopDate,"unixepoch") as stopped,
                (SELECT COUNT(uuid)
                 FROM TMTask AS PROJECT_TASK
                 WHERE
                   PROJECT_TASK.project = TASK.uuid AND
                   PROJECT_TASK.{self.IS_NOT_TRASHED} AND
                   PROJECT_TASK.{self.IS_OPEN}
                ) AS size,
                CASE
                    WHEN TASK.type = 0 THEN 'task'
                    WHEN TASK.type = 1 THEN 'project'
                    WHEN TASK.type = 2 THEN 'heading'
                END AS type
            FROM
                {self.TABLE_TASK} AS TASK
            LEFT OUTER JOIN
                {self.TABLE_TASK} PROJECT ON TASK.project = PROJECT.uuid
            LEFT OUTER JOIN
                {self.TABLE_AREA} AREA ON TASK.area = AREA.uuid
            LEFT OUTER JOIN
                {self.TABLE_TASK} HEADING ON TASK.actionGroup = HEADING.uuid
            LEFT OUTER JOIN
                {self.TABLE_TASK} HEADPROJ ON HEADING.project = HEADPROJ.uuid
            LEFT OUTER JOIN
                {self.TABLE_TASKTAG} TAGS ON TASK.uuid = TAGS.tasks
            LEFT OUTER JOIN
                {self.TABLE_TAG} TAG ON TAGS.tags = TAG.uuid
            WHERE
                {self.filter}
                {sql}
                """

        return self.execute_query(sql)

    def execute_query(self, sql):
        """Run the actual query"""
        if self.debug is True:
            print(sql)
        try:
            connection = sqlite3.connect(self.database)
            connection.row_factory = Things3.dict_factory
            cursor = connection.cursor()
            cursor.execute(sql)
            tasks = cursor.fetchall()
            tasks = self.anonymize_tasks(tasks)
            if self.debug:
                for task in tasks:
                    print(task)
            return tasks
        except sqlite3.OperationalError as error:
            print(f"Could not query the database at: {self.database}.")
            print(f"Details: {error}.")
            sys.exit(2)

    # pylint: disable=C0103
    def mode_project(self):
        """Hack to switch to project view"""
        self.IS_TASK = self.MODE_PROJECT

    # pylint: disable=C0103
    def mode_task(self):
        """Hack to switch to project view"""
        self.IS_TASK = self.MODE_TASK

    functions = {
        "inbox": get_inbox,
        "today": get_today,
        "next": get_anytime,
        "backlog": get_someday,
        "upcoming": get_upcoming,
        "waiting": get_waiting,
        "mit": get_mit,
        "completed": get_completed,
        "cancelled": get_cancelled,
        "trashed": get_trashed,
        "projects": get_projects,
        "areas": get_areas,
        "all": get_all,
        "due": get_due,
        "lint": get_lint,
        "empty": get_empty_projects,
        "cleanup": get_cleanup,
        "top-proj": get_largest_projects,
        "stats-day": get_daystats,
        "stats-min-today": get_minutes_today
    }


1			#!/usr/bin/env python3
2			# -- coding: utf-8 --
3
4			"""Simple read-only API for Things 3."""
5
6			from __future__ import print_function
7
8			__author__ = "Alexander Willner"
9			__copyright__ = "2020 Alexander Willner"
10			__credits__ = ["Alexander Willner"]
11			__license__ = "Apache License 2.0"
12			__version__ = "2.5.0.dev0"
13			__maintainer__ = "Alexander Willner"
14			__email__ = "[email protected]"
15			__status__ = "Development"
16
17			import sqlite3
18			import sys
19			from random import shuffle
20			from os import environ
21			import getpass
22
23
24			# pylint: disable=R0904
25			class Things3():
26			"""Simple read-only API for Things 3."""
27			# Variables
28			debug = False
29			database = None
30			json = False
31			filter = ""
32			tag_waiting = "Waiting" if not environ.get('TAG_WAITING') \
33			else environ.get('TAG_WAITING')
34			tag_mit = "MIT" if not environ.get('TAG_MIT') \
35			else environ.get('TAG_MIT')
36			tag_cleanup = "Cleanup" if not environ.get('TAG_CLEANUP') \
37			else environ.get('TAG_CLEANUP')
38			anonymize = bool(environ.get('ANONYMIZE'))
39
40			# Database info
41			FILE_DB = '/Library/Containers/'\
42			'com.culturedcode.ThingsMac/Data/Library/'\
43			'Application Support/Cultured Code/Things/Things.sqlite3'
44			FILE_SQLITE = '/Users/' + getpass.getuser() + FILE_DB \
45			if not environ.get('THINGSDB') else environ.get('THINGSDB')
46
47			TABLE_TASK = "TMTask"
48			TABLE_AREA = "TMArea"
49			TABLE_TAG = "TMTag"
50			TABLE_TASKTAG = "TMTaskTag"
51			DATE_CREATE = "creationDate"
52			DATE_MOD = "userModificationDate"
53			DATE_DUE = "dueDate"
54			DATE_START = "startDate"
55			DATE_STOP = "stopDate"
56			IS_INBOX = "start = 0"
57			IS_ANYTIME = "start = 1"
58			IS_SOMEDAY = "start = 2"
59			IS_SCHEDULED = f"{DATE_START} IS NOT NULL"
60			IS_NOT_SCHEDULED = f"{DATE_START} IS NULL"
61			IS_DUE = f"{DATE_DUE} IS NOT NULL"
62			IS_NOT_DUE = f"{DATE_DUE} IS NULL"
63			IS_RECURRING = "recurrenceRule IS NOT NULL"
64			IS_NOT_RECURRING = "recurrenceRule IS NULL"
65			IS_TASK = "type = 0"
66			IS_PROJECT = "type = 1"
67			IS_HEADING = "type = 2"
68			IS_TRASHED = "trashed = 1"
69			IS_NOT_TRASHED = "trashed = 0"
70			IS_OPEN = "status = 0"
71			IS_CANCELLED = "status = 2"
72			IS_DONE = "status = 3"
73			MODE_TASK = "type = 0"
74			MODE_PROJECT = "type = 1"
75
76			def __init__(self,
77			database=FILE_SQLITE,
78			tag_waiting='Waiting',
79			tag_mit='MIT',
80			json=False):
81			self.database = database if database is not None else self.FILE_SQLITE
82			self.tag_mit = tag_mit
83			self.tag_waiting = tag_waiting
84			self.json = json
85
86			@staticmethod
87			def anonymize_string(string):
88			"""Scramble text."""
89			if string is None:
90			return None
91			string = list(string)
92			shuffle(string)
93			string = ''.join(string)
94			return string
95
96			@staticmethod
97			def dict_factory(cursor, row):
98			"""Convert SQL result into a dictionary"""
99			dictionary = {}
100			for idx, col in enumerate(cursor.description):
101			dictionary[col[0]] = row[idx]
102			return dictionary
103
104			def anonymize_tasks(self, tasks):
105			"""Scramble output for screenshots."""
106			if self.anonymize:
107			for task in tasks:
108			task['title'] = self.anonymize_string(task['title'])
109			task['context'] = self.anonymize_string(task['context'])
110			return tasks
111
112			def get_inbox(self):
113			"""Get all tasks from the inbox."""
114			query = f"""
115			TASK.{self.IS_NOT_TRASHED} AND
116			TASK.{self.IS_TASK} AND
117			TASK.{self.IS_OPEN} AND
118			TASK.{self.IS_INBOX}
119			ORDER BY TASK.duedate DESC , TASK.todayIndex
120			"""
121			return self.get_rows(query)
122
123			def get_today(self):
124			"""Get all tasks from the today list."""
125			query = f"""
126			TASK.{self.IS_NOT_TRASHED} AND
127			TASK.{self.IS_TASK} AND
128			TASK.{self.IS_OPEN} AND
129			(TASK.{self.IS_ANYTIME} OR (
130			TASK.{self.IS_SOMEDAY} AND
131			TASK.{self.DATE_START} <= strftime('%s', 'now')
132			)
133			) AND
134			TASK.{self.IS_SCHEDULED} AND (
135			(
136			PROJECT.title IS NULL OR (
137			PROJECT.{self.IS_NOT_TRASHED}
138			)
139			) AND (
140			HEADPROJ.title IS NULL OR (
141			HEADPROJ.{self.IS_NOT_TRASHED}
142			)
143			)
144			)
145			ORDER BY TASK.duedate DESC , TASK.todayIndex
146			"""
147			return self.get_rows(query)
148
149			def get_someday(self):
150			"""Get someday tasks."""
151			query = f"""
152			TASK.{self.IS_NOT_TRASHED} AND
153			TASK.{self.IS_TASK} AND
154			TASK.{self.IS_OPEN} AND
155			TASK.{self.IS_SOMEDAY} AND
156			TASK.{self.IS_NOT_SCHEDULED} AND
157			TASK.{self.IS_NOT_RECURRING} AND (
158			(
159			PROJECT.title IS NULL OR (
160			PROJECT.{self.IS_NOT_TRASHED}
161			)
162			) AND (
163			HEADPROJ.title IS NULL OR (
164			HEADPROJ.{self.IS_NOT_TRASHED}
165			)
166			)
167			)
168			ORDER BY TASK.duedate DESC, TASK.creationdate DESC
169			"""
170			return self.get_rows(query)
171
172			def get_upcoming(self):
173			"""Get upcoming tasks."""
174			query = f"""
175			TASK.{self.IS_NOT_TRASHED} AND
176			TASK.{self.IS_TASK} AND
177			TASK.{self.IS_OPEN} AND
178			TASK.{self.IS_SOMEDAY} AND
179			TASK.{self.IS_SCHEDULED} AND
180			TASK.{self.IS_NOT_RECURRING} AND (
181			(
182			PROJECT.title IS NULL OR (
183			PROJECT.{self.IS_NOT_TRASHED}
184			)
185			) AND (
186			HEADPROJ.title IS NULL OR (
187			HEADPROJ.{self.IS_NOT_TRASHED}
188			)
189			)
190			)
191			ORDER BY TASK.startdate, TASK.todayIndex
192			"""
193			return self.get_rows(query)
194
195			def get_waiting(self):
196			"""Get waiting tasks."""
197			return self.get_tag(self.tag_waiting)
198
199			def get_mit(self):
200			"""Get most important tasks."""
201			return self.get_tag(self.tag_mit)
202
203			def get_tag(self, tag):
204			"""Get task with specific tag"""
205			query = f"""
206			TASK.{self.IS_NOT_TRASHED} AND
207			TASK.{self.IS_TASK} AND
208			TASK.{self.IS_OPEN} AND
209			TASK.{self.IS_NOT_RECURRING} AND
210			TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
211			WHERE title='{tag}'
212			)
213			AND (
214			(
215			PROJECT.title IS NULL OR (
216			PROJECT.{self.IS_NOT_TRASHED}
217			)
218			) AND (
219			HEADPROJ.title IS NULL OR (
220			HEADPROJ.{self.IS_NOT_TRASHED}
221			)
222			)
223			)
224			ORDER BY TASK.duedate DESC , TASK.todayIndex
225			"""
226			return self.get_rows(query)
227
228			def get_anytime(self):
229			"""Get anytime tasks."""
230			query = f"""
231			TASK.{self.IS_NOT_TRASHED} AND
232			TASK.{self.IS_TASK} AND
233			TASK.{self.IS_OPEN} AND
234			TASK.{self.IS_ANYTIME} AND
235			TASK.{self.IS_NOT_SCHEDULED} AND (
236			(
237			PROJECT.title IS NULL OR (
238			PROJECT.{self.IS_ANYTIME} AND
239			PROJECT.{self.IS_NOT_SCHEDULED} AND
240			PROJECT.{self.IS_NOT_TRASHED}
241			)
242			) AND (
243			HEADPROJ.title IS NULL OR (
244			HEADPROJ.{self.IS_ANYTIME} AND
245			HEADPROJ.{self.IS_NOT_SCHEDULED} AND
246			HEADPROJ.{self.IS_NOT_TRASHED}
247			)
248			)
249			)
250			ORDER BY TASK.duedate DESC , TASK.todayIndex
251			"""
252			if self.filter:
253			# ugly hack for Kanban task view on project
254			query = f"""
255			TASK.{self.IS_NOT_TRASHED} AND
256			TASK.{self.IS_TASK} AND
257			TASK.{self.IS_OPEN} AND
258			TASK.{self.IS_ANYTIME} AND
259			TASK.{self.IS_NOT_SCHEDULED} AND (
260			(
261			PROJECT.title IS NULL OR (
262			PROJECT.{self.IS_NOT_TRASHED}
263			)
264			) AND (
265			HEADPROJ.title IS NULL OR (
266			HEADPROJ.{self.IS_NOT_TRASHED}
267			)
268			)
269			)
270			ORDER BY TASK.duedate DESC , TASK.todayIndex
271			"""
272			return self.get_rows(query)
273
274			def get_completed(self):
275			"""Get completed tasks."""
276			query = f"""
277			TASK.{self.IS_NOT_TRASHED} AND
278			TASK.{self.IS_TASK} AND
279			TASK.{self.IS_DONE}
280			ORDER BY TASK.{self.DATE_STOP}
281			"""
282			return self.get_rows(query)
283
284			def get_cancelled(self):
285			"""Get cancelled tasks."""
286			query = f"""
287			TASK.{self.IS_NOT_TRASHED} AND
288			TASK.{self.IS_TASK} AND
289			TASK.{self.IS_CANCELLED}
290			ORDER BY TASK.{self.DATE_STOP}
291			"""
292			return self.get_rows(query)
293
294			def get_trashed(self):
295			"""Get trashed tasks."""
296			query = f"""
297			TASK.{self.IS_TRASHED} AND
298			TASK.{self.IS_TASK}
299			ORDER BY TASK.{self.DATE_STOP}
300			"""
301			return self.get_rows(query)
302
303			def get_projects(self):
304			"""Get projects."""
305			query = f"""
306			SELECT
307			TASK.uuid,
308			TASK.title,
309			NULL as context
310			FROM
311			{self.TABLE_TASK} AS TASK
312			WHERE
313			TASK.{self.IS_NOT_TRASHED} AND
314			TASK.{self.IS_PROJECT} AND
315			TASK.{self.IS_OPEN}
316			ORDER BY TASK.title
317			"""
318			return self.execute_query(query)
319
320			def get_areas(self):
321			"""Get areas."""
322			query = f"""
323			SELECT
324			AREA.uuid AS uuid,
325			AREA.title AS title
326			FROM
327			{self.TABLE_AREA} AS AREA
328			ORDER BY AREA.title
329			"""
330			return self.execute_query(query)
331
332			def get_all(self):
333			"""Get all tasks."""
334			query = f"""
335			TASK.{self.IS_NOT_TRASHED} AND
336			TASK.{self.IS_TASK} AND (
337			(
338			PROJECT.title IS NULL OR (
339			PROJECT.{self.IS_NOT_TRASHED}
340			)
341			) AND (
342			HEADPROJ.title IS NULL OR (
343			HEADPROJ.{self.IS_NOT_TRASHED}
344			)
345			)
346			)
347			"""
348			return self.get_rows(query)
349
350			def get_due(self):
351			"""Get due tasks."""
352			query = f"""
353			TASK.{self.IS_NOT_TRASHED} AND
354			TASK.{self.IS_TASK} AND
355			TASK.{self.IS_OPEN} AND
356			TASK.{self.IS_DUE} AND (
357			(
358			PROJECT.title IS NULL OR (
359			PROJECT.{self.IS_NOT_TRASHED}
360			)
361			) AND (
362			HEADPROJ.title IS NULL OR (
363			HEADPROJ.{self.IS_NOT_TRASHED}
364			)
365			)
366			)
367			ORDER BY TASK.{self.DATE_DUE}
368			"""
369			return self.get_rows(query)
370
371			def get_lint(self):
372			"""Get tasks that float around"""
373			query = f"""
374			TASK.{self.IS_NOT_TRASHED} AND
375			TASK.{self.IS_OPEN} AND
376			TASK.{self.IS_TASK} AND
377			(TASK.{self.IS_SOMEDAY} OR TASK.{self.IS_ANYTIME}) AND
378			TASK.project IS NULL AND
379			TASK.area IS NULL AND
380			TASK.actionGroup IS NULL
381			"""
382			return self.get_rows(query)
383
384			def get_empty_projects(self):
385			"""Get projects that are empty"""
386			query = f"""
387			TASK.{self.IS_NOT_TRASHED} AND
388			TASK.{self.IS_OPEN} AND
389			TASK.{self.IS_PROJECT} AND
390			(TASK.{self.IS_ANYTIME} OR TASK.{self.IS_SCHEDULED})
391			GROUP BY TASK.uuid
392			HAVING
393			(SELECT COUNT(uuid)
394			FROM TMTask AS PROJECT_TASK
395			WHERE
396			PROJECT_TASK.project = TASK.uuid AND
397			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
398			PROJECT_TASK.{self.IS_OPEN} AND
399			PROJECT_TASK.{self.IS_ANYTIME}
400			) = 0
401			"""
402			return self.get_rows(query)
403
404			def get_largest_projects(self):
405			"""Get projects that are empty"""
406			query = f"""
407			SELECT
408			TASK.uuid,
409			TASK.title AS title,
410			creationDate AS created,
411			userModificationDate AS modified,
412			(SELECT COUNT(uuid)
413			FROM TMTask AS PROJECT_TASK
414			WHERE
415			PROJECT_TASK.project = TASK.uuid AND
416			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
417			PROJECT_TASK.{self.IS_OPEN}
418			) AS tasks
419			FROM
420			{self.TABLE_TASK} AS TASK
421			WHERE
422			TASK.{self.IS_NOT_TRASHED} AND
423			TASK.{self.IS_OPEN} AND
424			TASK.{self.IS_PROJECT}
425			GROUP BY TASK.uuid
426			ORDER BY tasks DESC
427			"""
428			return self.execute_query(query)
429
430			def get_daystats(self):
431			"""Get a history of task activities"""
432			days = 365
433			query = f"""
434			WITH RECURSIVE timeseries(x) AS (
435			SELECT 0
436			UNION ALL
437			SELECT x+1 FROM timeseries
438			LIMIT {days}
439			)
440			SELECT
441			date(julianday("now", "-{days} days"),
442			"+" \|\| x \|\| " days") as date,
443			CREATED.TasksCreated as created,
444			CLOSED.TasksClosed as completed,
445			CANCELLED.TasksCancelled as cancelled,
446			TRASHED.TasksTrashed as trashed
447			FROM timeseries
448			LEFT JOIN
449			(SELECT COUNT(uuid) AS TasksCreated,
450			date(creationDate,"unixepoch") AS DAY
451			FROM {self.TABLE_TASK} AS TASK
452			WHERE DAY NOT NULL
453			AND TASK.{self.IS_TASK}
454			GROUP BY DAY)
455			AS CREATED ON CREATED.DAY = date
456			LEFT JOIN
457			(SELECT COUNT(uuid) AS TasksCancelled,
458			date(stopDate,"unixepoch") AS DAY
459			FROM {self.TABLE_TASK} AS TASK
460			WHERE DAY NOT NULL
461			AND TASK.{self.IS_CANCELLED} AND TASK.{self.IS_TASK}
462			GROUP BY DAY)
463			AS CANCELLED ON CANCELLED.DAY = date
464			LEFT JOIN
465			(SELECT COUNT(uuid) AS TasksTrashed,
466			date(userModificationDate,"unixepoch") AS DAY
467			FROM {self.TABLE_TASK} AS TASK
468			WHERE DAY NOT NULL
469			AND TASK.{self.IS_TRASHED} AND TASK.{self.IS_TASK}
470			GROUP BY DAY)
471			AS TRASHED ON TRASHED.DAY = date
472			LEFT JOIN
473			(SELECT COUNT(uuid) AS TasksClosed,
474			date(stopDate,"unixepoch") AS DAY
475			FROM {self.TABLE_TASK} AS TASK
476			WHERE DAY NOT NULL
477			AND TASK.{self.IS_DONE} AND TASK.{self.IS_TASK}
478			GROUP BY DAY)
479			AS CLOSED ON CLOSED.DAY = date
480			"""
481			return self.execute_query(query)
482
483			def get_minutes_today(self):
484			"""Count the planned minutes for today."""
485			query = f"""
486			SELECT
487			SUM(TAG.title) AS minutes
488			FROM
489			{self.TABLE_TASK} AS TASK
490			LEFT OUTER JOIN
491			TMTask PROJECT ON TASK.project = PROJECT.uuid
492			LEFT OUTER JOIN
493			TMArea AREA ON TASK.area = AREA.uuid
494			LEFT OUTER JOIN
495			TMTask HEADING ON TASK.actionGroup = HEADING.uuid
496			LEFT OUTER JOIN
497			TMTask HEADPROJ ON HEADING.project = HEADPROJ.uuid
498			LEFT OUTER JOIN
499			TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
500			LEFT OUTER JOIN
501			TMTag TAG ON TAGS.tags = TAG.uuid
502			WHERE
503			printf("%d", TAG.title) = TAG.title AND
504			TASK.{self.IS_NOT_TRASHED} AND
505			TASK.{self.IS_TASK} AND
506			TASK.{self.IS_OPEN} AND
507			TASK.{self.IS_ANYTIME} AND
508			TASK.{self.IS_SCHEDULED} AND (
509			(
510			PROJECT.title IS NULL OR (
511			PROJECT.{self.IS_NOT_TRASHED}
512			)
513			) AND (
514			HEADPROJ.title IS NULL OR (
515			HEADPROJ.{self.IS_NOT_TRASHED}
516			)
517			)
518			)
519			"""
520			return self.execute_query(query)
521
522			def get_cleanup(self):
523			"""Tasks and projects that need work."""
524			result = []
525			result.extend(self.get_lint())
526			result.extend(self.get_empty_projects())
527			result.extend(self.get_tag(self.tag_cleanup))
528			return result
529
530			@staticmethod
531			def get_not_implemented():
532			"""Not implemented warning."""
533			return [{"title": "not implemented"}]
534
535			def get_rows(self, sql):
536			"""Query Things database."""
537
538			sql = f"""
539			SELECT DISTINCT
540			TASK.uuid,
541			TASK.title,
542			CASE
543			WHEN AREA.title IS NOT NULL THEN AREA.title
544			WHEN PROJECT.title IS NOT NULL THEN PROJECT.title
545			WHEN HEADING.title IS NOT NULL THEN HEADING.title
546			END AS context,
547			CASE
548			WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
549			WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
550			END AS context_uuid,
551			CASE
552			WHEN TASK.recurrenceRule IS NULL
553			THEN date(TASK.dueDate,"unixepoch")
554			ELSE NULL
555			END AS due,
556			date(TASK.creationDate,"unixepoch") as created,
557			date(TASK.userModificationDate,"unixepoch") as modified,
558			date(TASK.startDate,"unixepoch") as started,
559			date(TASK.stopDate,"unixepoch") as stopped,
560			(SELECT COUNT(uuid)
561			FROM TMTask AS PROJECT_TASK
562			WHERE
563			PROJECT_TASK.project = TASK.uuid AND
564			PROJECT_TASK.{self.IS_NOT_TRASHED} AND
565			PROJECT_TASK.{self.IS_OPEN}
566			) AS size,
567			CASE
568			WHEN TASK.type = 0 THEN 'task'
569			WHEN TASK.type = 1 THEN 'project'
570			WHEN TASK.type = 2 THEN 'heading'
571			END AS type
572			FROM
573			{self.TABLE_TASK} AS TASK
574			LEFT OUTER JOIN
575			{self.TABLE_TASK} PROJECT ON TASK.project = PROJECT.uuid
576			LEFT OUTER JOIN
577			{self.TABLE_AREA} AREA ON TASK.area = AREA.uuid
578			LEFT OUTER JOIN
579			{self.TABLE_TASK} HEADING ON TASK.actionGroup = HEADING.uuid
580			LEFT OUTER JOIN
581			{self.TABLE_TASK} HEADPROJ ON HEADING.project = HEADPROJ.uuid
582			LEFT OUTER JOIN
583			{self.TABLE_TASKTAG} TAGS ON TASK.uuid = TAGS.tasks
584			LEFT OUTER JOIN
585			{self.TABLE_TAG} TAG ON TAGS.tags = TAG.uuid
586			WHERE
587			{self.filter}
588			{sql}
589			"""
590
591			return self.execute_query(sql)
592
593			def execute_query(self, sql):
594			"""Run the actual query"""
595			if self.debug is True:
596			print(sql)
597			try:
598			connection = sqlite3.connect(self.database)
599			connection.row_factory = Things3.dict_factory
600			cursor = connection.cursor()
601			cursor.execute(sql)
602			tasks = cursor.fetchall()
603			tasks = self.anonymize_tasks(tasks)
604			if self.debug:
605			for task in tasks:
606			print(task)
607			return tasks
608			except sqlite3.OperationalError as error:
609			print(f"Could not query the database at: {self.database}.")
610			print(f"Details: {error}.")
611			sys.exit(2)
612
613			# pylint: disable=C0103
614			def mode_project(self):
615			"""Hack to switch to project view"""
616			self.IS_TASK = self.MODE_PROJECT
617
618			# pylint: disable=C0103
619			def mode_task(self):
620			"""Hack to switch to project view"""
621			self.IS_TASK = self.MODE_TASK
622
623			functions = {
624			"inbox": get_inbox,
625			"today": get_today,
626			"next": get_anytime,
627			"backlog": get_someday,
628			"upcoming": get_upcoming,
629			"waiting": get_waiting,
630			"mit": get_mit,
631			"completed": get_completed,
632			"cancelled": get_cancelled,
633			"trashed": get_trashed,
634			"projects": get_projects,
635			"areas": get_areas,
636			"all": get_all,
637			"due": get_due,
638			"lint": get_lint,
639			"empty": get_empty_projects,
640			"cleanup": get_cleanup,
641			"top-proj": get_largest_projects,
642			"stats-day": get_daystats,
643			"stats-min-today": get_minutes_today
644			}
645

AlexanderWillner / KanbanView

Push — master ( dd0bda...e4c11f )

things3.things3.Things3.get_mit() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like