things3.things3.Things3.get_largest_projects() - Code Metrics - Inspection of "Merge pull request #12 from AlexanderWillner/fix-g..." - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e4c11f...a69919 )

by Alexander

created 2020-04-22 18:19 UTC

things3.things3.Things3.get_largest_projects() A

↳ Parent: things3.things3

Complexity

Conditions

Size

Total Lines	25
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import sqlite3
import sys
from random import shuffle
from os import environ
import getpass
import configparser
from pathlib import Path


# pylint: disable=R0904
class Things3():
    """Simple read-only API for Things 3."""

    # Database info
    FILE_DB = '/Library/Containers/'\
              'com.culturedcode.ThingsMac/Data/Library/'\
              'Application Support/Cultured Code/Things/Things.sqlite3'

    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"
    RECURRING_IS_NOT_PAUSED = "instanceCreationPaused = 0"
    RECURRING_HAS_NEXT_STARTDATE = "nextInstanceStartDate IS NOT NULL"
    MODE_TASK = "type = 0"
    MODE_PROJECT = "type = 1"

    # Variables
    debug = False
    user = getpass.getuser()
    database = f"/Users/{user}/{FILE_DB}"
    filter = ""
    tag_waiting = "Waiting"
    tag_mit = "MIT"
    tag_cleanup = "Cleanup"
    anonymize = False
    config = configparser.ConfigParser()
    config.read(str(Path.home()) + '/.kanbanviewrc')

    # pylint: disable=R0913
    def __init__(self,
                 database=None,
                 tag_waiting=None,
                 tag_mit=None,
                 tag_cleanup=None,
                 anonymize=None):

        cfg = self.get_from_config(self.config, tag_waiting, 'TAG_WAITING')
        self.tag_waiting = cfg if cfg else self.tag_waiting

        cfg = self.get_from_config(self.config, anonymize, 'ANONYMIZE')
        self.anonymize = cfg if cfg else self.anonymize

        cfg = self.get_from_config(self.config, tag_mit, 'TAG_MIT')
        self.tag_mit = cfg if cfg else self.tag_mit

        cfg = self.get_from_config(self.config, tag_cleanup, 'TAG_CLEANUP')
        self.tag_cleanup = cfg if cfg else self.tag_cleanup

        cfg = self.get_from_config(self.config, database, 'THINGSDB')
        self.database = cfg if cfg else self.database

    @staticmethod
    def get_from_config(config, variable, preference):
        """Set variable. Priority: input, environment, config"""
        result = None
        if variable is not None:
            result = variable
        elif environ.get(preference):
            result = environ.get(preference)
        elif 'DATABASE' in config and preference in config['DATABASE']:
            result = config['DATABASE'][preference]
        return result

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

AlexanderWillner / KanbanView

Push — master ( e4c11f...a69919 )

things3.things3.Things3.get_largest_projects() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like