things3.things3.Things3.anonymize_string() - Code Metrics - Inspection of "Merge branch 'master' of https://github.com/Alexan..." - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 764061...ef0dcc )

by Alexander

created 2020-04-07 09:43 UTC

things3.things3.Things3.anonymize_string() A

↳ Parent: things3.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
    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')
    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.json = json

    @staticmethod
    def anonymize_string(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_string(str(task[self.I_TITLE]))
                task[self.I_CONTEXT] = \
                    self.anonymize_string(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

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

AlexanderWillner / KanbanView

Push — master ( 764061...ef0dcc )

things3.things3.Things3.anonymize_string() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like