src.things3.Things3.get_waiting() - Code Metrics - Inspection of "moved and refactored method 'anonymize'" - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b6cc82...5a7423 )

by Alexander

created 2020-04-06 12:51 UTC

src.things3.Things3.get_waiting() A

↳ Parent: src.things3

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	1
dl	0
loc	7
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"
__version__ = "2.0.0"
__maintainer__ = "Alexander Willner"
__email__ = "[email protected]"
__status__ = "Development"

import sqlite3
from random import shuffle
from os.path import expanduser
from os import environ


class Things3():
    """Simple read-only API for Things 3."""
    # Variables
    database = None
    cursor = None
    json = False
    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')

    # Basic config
    ANONYMIZE = bool(environ.get('ANONYMIZE'))

    # Database info
    FILE_SQLITE = '~/Library/Containers/'\
                  'com.culturedcode.ThingsMac.beta/Data/Library/'\
                  'Application Support/Cultured Code/Things/Things.sqlite3'\
        if not environ.get('THINGSDB') else environ.get('THINGSDB')
    TASKTABLE = "TMTask"
    AREATABLE = "TMArea"
    TAGTABLE = "TMTag"
    TASKTAGTABLE = "TMTaskTag"
    ISNOTTRASHED = "TASK.trashed = 0"
    ISTRASHED = "TASK.trashed = 1"
    ISOPEN = "TASK.status = 0"
    ISNOTSTARTED = "TASK.start = 0"
    ISCANCELLED = "TASK.status = 2"
    ISCOMPLETED = "TASK.status = 3"
    ISSTARTED = "TASK.start = 1"
    ISPOSTPONED = "TASK.start = 2"
    ISTASK = "TASK.type = 0"
    ISPROJECT = "TASK.type = 1"
    ISHEADING = "TASK.type = 2"
    ISOPENTASK = ISTASK + " AND " + ISNOTTRASHED + " AND " + ISOPEN
    DATECREATE = "creationDate"
    DATEMOD = "userModificationDate"
    DATEDUE = "dueDate"
    DATESTART = "unixepoch"
    DATESTOP = "stopDate"

    # Query Index
    I_UUID = 0
    I_TITLE = 1
    I_CONTEXT = 2
    I_CONTEXT_UUID = 3
    I_DUE = 4
    I_CREATE = 5
    I_MOD = 6
    I_START = 7
    I_STOP = 8

    def __init__(self,
                 database=FILE_SQLITE,
                 tag_waiting='Waiting',
                 tag_mit='MIT',
                 json=False):
        self.database = expanduser(database)
        self.tag_mit = tag_mit
        self.tag_waiting = tag_waiting
        self.cursor = sqlite3.connect(self.database).cursor()
        self.json = json

    @staticmethod
    def anonymize(string):
        """Scramble text."""
        string = list(string)
        shuffle(string)
        string = ''.join(string)
        return string

    def anonymize_tasks(self, tasks):
        """Scramble output for screenshots."""
        result = tasks
        if self.ANONYMIZE:
            result = []
            for task in tasks:
                task = list(task)
                task[self.I_TITLE] = self.anonymize(str(task[self.I_TITLE]))
                task[self.I_CONTEXT] = \
                    self.anonymize(str(task[self.I_CONTEXT]))
                result.append(task)
        return result

    def get_inbox(self):
        """Get all tasks from the inbox."""
        query = self.ISOPENTASK + " AND " + self.ISNOTSTARTED + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_today(self):
        """Get all tasks from the today list."""
        query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
            " AND TASK.startdate is NOT NULL" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_someday(self):
        """Get someday tasks."""
        query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
            " AND TASK.startdate IS NULL AND TASK.recurrenceRule IS NULL" + \
            " ORDER BY TASK.duedate DESC, TASK.creationdate DESC"
        return self.get_rows(query)

    def get_upcoming(self):
        """Get upcoming tasks."""
        query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
            " AND (TASK.startDate NOT NULL " + \
            "      OR TASK.recurrenceRule NOT NULL)" + \
            " ORDER BY TASK.startdate, TASK.todayIndex"
        return self.get_rows(query)

    def get_waiting(self):
        """Get waiting tasks."""
        query = self.ISOPENTASK + \
            " AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
            " WHERE title='" + self.tag_waiting + "')" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_mit(self):
        """Get most important tasks."""
        query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
            " AND PROJECT.status = 0 " \
            " AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
            " WHERE title='" + self.tag_mit + "')" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_anytime(self):
        """Get anytime tasks."""
        query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
            " AND TASK.startdate is NULL" + \
            " AND (TASK.area NOT NULL OR TASK.project in " + \
            "(SELECT uuid FROM " + self.TASKTABLE + \
            " WHERE uuid=TASK.project AND start=1" + \
            " AND trashed=0))" + \
            " ORDER BY TASK.duedate DESC , TASK.todayIndex"
        return self.get_rows(query)

    def get_completed(self):
        """Get completed tasks."""
        query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
            " AND " + self.ISCOMPLETED + \
            " ORDER BY TASK." + self.DATESTOP
        return self.get_rows(query)

    def get_cancelled(self):
        """Get cancelled tasks."""
        query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
            " AND " + self.ISCANCELLED + \
            " ORDER BY TASK." + self.DATESTOP
        return self.get_rows(query)

    def get_trashed(self):
        """Get trashed tasks."""
        query = self.ISTRASHED + " AND " + self.ISTASK + \
            " ORDER BY TASK." + self.DATESTOP
        return self.get_rows(query)

    def get_all(self):
        """Get all tasks."""
        query = self.ISNOTTRASHED + " AND " + self.ISTASK
        return self.get_rows(query)

    def get_due(self):
        """Get due tasks."""
        query = self.ISOPENTASK + " AND TASK.dueDate NOT NULL" + \
            " ORDER BY TASK.dueDate"
        return self.get_rows(query)

    @staticmethod
    def get_not_implemented():
        """Not implemented warning."""
        return [["0", "not implemented", "no context", "0", "0", "0", "0",
                 "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0"]]

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

        sql = """
            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,
                CASE
                    WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
                    WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
                END,
                CASE
                    WHEN TASK.recurrenceRule IS NULL
                    THEN date(TASK.dueDate,"unixepoch")
                ELSE NULL
                END,
                date(TASK.creationDate,"unixepoch"),
                date(TASK.userModificationDate,"unixepoch"),
                date(TASK.startDate,"unixepoch"),
                date(TASK.stopDate,"unixepoch")
            FROM
                TMTask 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
                TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
            LEFT OUTER JOIN
                TMTag TAG ON TAGS.tags = TAG.uuid
            WHERE """ + sql

        self.cursor.execute(sql)
        tasks = self.cursor.fetchall()
        tasks = self.anonymize_tasks(tasks)

        return tasks

    def convert_task_to_model(self, task):
        """Convert task to model."""
        model = {'uuid': task[self.I_UUID],
                 'title': task[self.I_TITLE],
                 'context': task[self.I_CONTEXT],
                 'context_uuid': task[self.I_CONTEXT_UUID],
                 'due': task[self.I_DUE],
                 'created': task[self.I_CREATE],
                 'modified': task[self.I_MOD],
                 'started': task[self.I_START],
                 'stopped': task[self.I_STOP]
                 }
        return model

    def convert_tasks_to_model(self, tasks):
        """Convert tasks to model."""
        model = []
        for task in tasks:
            model.append(self.convert_task_to_model(task))
        return model

    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,
        "all": get_all,
        "due": get_due,
    }


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"
12			__version__ = "2.0.0"
13			__maintainer__ = "Alexander Willner"
14			__email__ = "[email protected]"
15			__status__ = "Development"
16
17			import sqlite3
18			from random import shuffle
19			from os.path import expanduser
20			from os import environ
21
22
23			class Things3():
24			"""Simple read-only API for Things 3."""
25			# Variables
26			database = None
27			cursor = None
28			json = False
29			tag_waiting = "Waiting" if not environ.get('TAG_WAITING') \
30			else environ.get('TAG_WAITING')
31			tag_mit = "MIT" if not environ.get('TAG_MIT') \
32			else environ.get('TAG_MIT')
33
34			# Basic config
35			ANONYMIZE = bool(environ.get('ANONYMIZE'))
36
37			# Database info
38			FILE_SQLITE = '~/Library/Containers/'\
39			'com.culturedcode.ThingsMac.beta/Data/Library/'\
40			'Application Support/Cultured Code/Things/Things.sqlite3'\
41			if not environ.get('THINGSDB') else environ.get('THINGSDB')
42			TASKTABLE = "TMTask"
43			AREATABLE = "TMArea"
44			TAGTABLE = "TMTag"
45			TASKTAGTABLE = "TMTaskTag"
46			ISNOTTRASHED = "TASK.trashed = 0"
47			ISTRASHED = "TASK.trashed = 1"
48			ISOPEN = "TASK.status = 0"
49			ISNOTSTARTED = "TASK.start = 0"
50			ISCANCELLED = "TASK.status = 2"
51			ISCOMPLETED = "TASK.status = 3"
52			ISSTARTED = "TASK.start = 1"
53			ISPOSTPONED = "TASK.start = 2"
54			ISTASK = "TASK.type = 0"
55			ISPROJECT = "TASK.type = 1"
56			ISHEADING = "TASK.type = 2"
57			ISOPENTASK = ISTASK + " AND " + ISNOTTRASHED + " AND " + ISOPEN
58			DATECREATE = "creationDate"
59			DATEMOD = "userModificationDate"
60			DATEDUE = "dueDate"
61			DATESTART = "unixepoch"
62			DATESTOP = "stopDate"
63
64			# Query Index
65			I_UUID = 0
66			I_TITLE = 1
67			I_CONTEXT = 2
68			I_CONTEXT_UUID = 3
69			I_DUE = 4
70			I_CREATE = 5
71			I_MOD = 6
72			I_START = 7
73			I_STOP = 8
74
75			def __init__(self,
76			database=FILE_SQLITE,
77			tag_waiting='Waiting',
78			tag_mit='MIT',
79			json=False):
80			self.database = expanduser(database)
81			self.tag_mit = tag_mit
82			self.tag_waiting = tag_waiting
83			self.cursor = sqlite3.connect(self.database).cursor()
84			self.json = json
85
86			@staticmethod
87			def anonymize(string):
88			"""Scramble text."""
89			string = list(string)
90			shuffle(string)
91			string = ''.join(string)
92			return string
93
94			def anonymize_tasks(self, tasks):
95			"""Scramble output for screenshots."""
96			result = tasks
97			if self.ANONYMIZE:
98			result = []
99			for task in tasks:
100			task = list(task)
101			task[self.I_TITLE] = self.anonymize(str(task[self.I_TITLE]))
102			task[self.I_CONTEXT] = \
103			self.anonymize(str(task[self.I_CONTEXT]))
104			result.append(task)
105			return result
106
107			def get_inbox(self):
108			"""Get all tasks from the inbox."""
109			query = self.ISOPENTASK + " AND " + self.ISNOTSTARTED + \
110			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
111			return self.get_rows(query)
112
113			def get_today(self):
114			"""Get all tasks from the today list."""
115			query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
116			" AND TASK.startdate is NOT NULL" + \
117			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
118			return self.get_rows(query)
119
120			def get_someday(self):
121			"""Get someday tasks."""
122			query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
123			" AND TASK.startdate IS NULL AND TASK.recurrenceRule IS NULL" + \
124			" ORDER BY TASK.duedate DESC, TASK.creationdate DESC"
125			return self.get_rows(query)
126
127			def get_upcoming(self):
128			"""Get upcoming tasks."""
129			query = self.ISOPENTASK + " AND " + self.ISPOSTPONED + \
130			" AND (TASK.startDate NOT NULL " + \
131			" OR TASK.recurrenceRule NOT NULL)" + \
132			" ORDER BY TASK.startdate, TASK.todayIndex"
133			return self.get_rows(query)
134
135			def get_waiting(self):
136			"""Get waiting tasks."""
137			query = self.ISOPENTASK + \
138			" AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
139			" WHERE title='" + self.tag_waiting + "')" + \
140			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
141			return self.get_rows(query)
142
143			def get_mit(self):
144			"""Get most important tasks."""
145			query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
146			" AND PROJECT.status = 0 " \
147			" AND TAGS.tags=(SELECT uuid FROM " + self.TAGTABLE + \
148			" WHERE title='" + self.tag_mit + "')" + \
149			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
150			return self.get_rows(query)
151
152			def get_anytime(self):
153			"""Get anytime tasks."""
154			query = self.ISOPENTASK + " AND " + self.ISSTARTED + \
155			" AND TASK.startdate is NULL" + \
156			" AND (TASK.area NOT NULL OR TASK.project in " + \
157			"(SELECT uuid FROM " + self.TASKTABLE + \
158			" WHERE uuid=TASK.project AND start=1" + \
159			" AND trashed=0))" + \
160			" ORDER BY TASK.duedate DESC , TASK.todayIndex"
161			return self.get_rows(query)
162
163			def get_completed(self):
164			"""Get completed tasks."""
165			query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
166			" AND " + self.ISCOMPLETED + \
167			" ORDER BY TASK." + self.DATESTOP
168			return self.get_rows(query)
169
170			def get_cancelled(self):
171			"""Get cancelled tasks."""
172			query = self.ISNOTTRASHED + " AND " + self.ISTASK + \
173			" AND " + self.ISCANCELLED + \
174			" ORDER BY TASK." + self.DATESTOP
175			return self.get_rows(query)
176
177			def get_trashed(self):
178			"""Get trashed tasks."""
179			query = self.ISTRASHED + " AND " + self.ISTASK + \
180			" ORDER BY TASK." + self.DATESTOP
181			return self.get_rows(query)
182
183			def get_all(self):
184			"""Get all tasks."""
185			query = self.ISNOTTRASHED + " AND " + self.ISTASK
186			return self.get_rows(query)
187
188			def get_due(self):
189			"""Get due tasks."""
190			query = self.ISOPENTASK + " AND TASK.dueDate NOT NULL" + \
191			" ORDER BY TASK.dueDate"
192			return self.get_rows(query)
193
194			@staticmethod
195			def get_not_implemented():
196			"""Not implemented warning."""
197			return [["0", "not implemented", "no context", "0", "0", "0", "0",
198			"0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0", "0"]]
199
200			def get_rows(self, sql):
201			"""Query Things database."""
202
203			sql = """
204			SELECT DISTINCT
205			TASK.uuid,
206			TASK.title,
207			CASE
208			WHEN AREA.title IS NOT NULL THEN AREA.title
209			WHEN PROJECT.title IS NOT NULL THEN PROJECT.title
210			WHEN HEADING.title IS NOT NULL THEN HEADING.title
211			END,
212			CASE
213			WHEN AREA.uuid IS NOT NULL THEN AREA.uuid
214			WHEN PROJECT.uuid IS NOT NULL THEN PROJECT.uuid
215			END,
216			CASE
217			WHEN TASK.recurrenceRule IS NULL
218			THEN date(TASK.dueDate,"unixepoch")
219			ELSE NULL
220			END,
221			date(TASK.creationDate,"unixepoch"),
222			date(TASK.userModificationDate,"unixepoch"),
223			date(TASK.startDate,"unixepoch"),
224			date(TASK.stopDate,"unixepoch")
225			FROM
226			TMTask AS TASK
227			LEFT OUTER JOIN
228			TMTask PROJECT ON TASK.project = PROJECT.uuid
229			LEFT OUTER JOIN
230			TMArea AREA ON TASK.area = AREA.uuid
231			LEFT OUTER JOIN
232			TMTask HEADING ON TASK.actionGroup = HEADING.uuid
233			LEFT OUTER JOIN
234			TMTaskTag TAGS ON TASK.uuid = TAGS.tasks
235			LEFT OUTER JOIN
236			TMTag TAG ON TAGS.tags = TAG.uuid
237			WHERE """ + sql
238
239			self.cursor.execute(sql)
240			tasks = self.cursor.fetchall()
241			tasks = self.anonymize_tasks(tasks)
242
243			return tasks
244
245			def convert_task_to_model(self, task):
246			"""Convert task to model."""
247			model = {'uuid': task[self.I_UUID],
248			'title': task[self.I_TITLE],
249			'context': task[self.I_CONTEXT],
250			'context_uuid': task[self.I_CONTEXT_UUID],
251			'due': task[self.I_DUE],
252			'created': task[self.I_CREATE],
253			'modified': task[self.I_MOD],
254			'started': task[self.I_START],
255			'stopped': task[self.I_STOP]
256			}
257			return model
258
259			def convert_tasks_to_model(self, tasks):
260			"""Convert tasks to model."""
261			model = []
262			for task in tasks:
263			model.append(self.convert_task_to_model(task))
264			return model
265
266			functions = {
267			"inbox": get_inbox,
268			"today": get_today,
269			"next": get_anytime,
270			"backlog": get_someday,
271			"upcoming": get_upcoming,
272			"waiting": get_waiting,
273			"mit": get_mit,
274			"completed": get_completed,
275			"cancelled": get_cancelled,
276			"trashed": get_trashed,
277			"all": get_all,
278			"due": get_due,
279			}
280

AlexanderWillner / KanbanView

Push — master ( b6cc82...5a7423 )

src.things3.Things3.get_waiting() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like