things3.things3.Things3.get_not_implemented() - Code Metrics - Inspection of "new GUI, more statistics, more tests, global proje..." - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2a7579...a2cb78 )

by Alexander

created 2020-04-21 19:59 UTC

things3.things3.Things3.get_not_implemented() A

↳ Parent: things3.things3

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	0
dl	0
loc	4
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.0dev"
__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"

    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
            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 toggle_mode(self):
        """Hack to switch to project view"""
        if self.IS_TASK == "type = 1":
            self.IS_TASK = "type = 0"
        else:
            self.IS_TASK = "type = 1"

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

AlexanderWillner / KanbanView

Push — master ( 2a7579...a2cb78 )

things3.things3.Things3.get_not_implemented() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like