things3.things3.Things3.__init__() - Code Metrics - Inspection of "fixed #20 (matrix configuration) + added other con..." - AlexanderWillner/KanbanView - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( db01a9...08f366 )

by Alexander

created 2020-05-04 17:55 UTC

things3.things3.Things3.init() C

↳ Parent: things3.things3

Complexity

Conditions

Size

Total Lines	51
Code Lines	41

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	41
nop	11
dl	0
loc	51
rs	5.4
c	0
b	0
f	0

How to fix Long Method Complexity Many Parameters

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like things3.things3.Things3.__init__() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

Many Parameters

Methods with many parameters are not only hard to understand, but their parameters also often become inconsistent when you need more, or different data.

There are several approaches to avoid long parameter lists:

If you can get the data from another object that the method knows about already: Replace the parameter with method call
If several parameters are part of another object: Preserve Whole Object
If parameters are not yet connected: Introduce Parameter Object

#!/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.6.0"
__maintainer__ = "Alexander Willner"
__email__ = "[email protected]"
__status__ = "Development"

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


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

    # Database info
    FILE_CONFIG = str(Path.home()) + '/.kanbanviewrc'
    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"
    tag_a = "A"
    tag_b = "B"
    tag_c = "C"
    tag_d = "D"
    stat_Days = 365
    anonymize = False
    config = configparser.ConfigParser()
    config.read(FILE_CONFIG)

    # pylint: disable=R0913
    def __init__(self,
                 database=None,
                 tag_waiting=None,
                 tag_mit=None,
                 tag_cleanup=None,
                 tag_a=None,
                 tag_b=None,
                 tag_c=None,
                 tag_d=None,
                 stat_days=None,
                 anonymize=None):

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

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

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

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

        cfg = self.get_from_config(tag_a, 'TAG_A')
        self.tag_a = cfg if cfg else self.tag_a
        self.set_config('TAG_A', self.tag_a)

        cfg = self.get_from_config(tag_b, 'TAG_B')
        self.tag_b = cfg if cfg else self.tag_b
        self.set_config('TAG_B', self.tag_b)

        cfg = self.get_from_config(tag_c, 'TAG_C')
        self.tag_c = cfg if cfg else self.tag_c
        self.set_config('TAG_C', self.tag_c)

        cfg = self.get_from_config(tag_d, 'TAG_D')
        self.tag_d = cfg if cfg else self.tag_d
        self.set_config('TAG_D', self.tag_d)

        cfg = self.get_from_config(stat_days, 'STAT_DAYS')
        self.stat_days = cfg if cfg else self.stat_days
        self.set_config('STAT_DAYS', self.stat_days)

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

    def set_config(self, key, value, domain='DATABASE'):
        """Write variable to config."""
        if domain not in self.config:
            self.config.add_section(domain)
        if value is not None and key is not None:
            self.config.set(domain, str(key), str(value))
            with open(self.FILE_CONFIG, "w+") as configfile:
                self.config.write(configfile)

    def get_config(self, key, domain='DATABASE'):
        """Get variable from config."""
        result = None
        if domain in self.config and key in self.config[domain]:
            result = path.expanduser(self.config[domain][key])
        return result

    def get_from_config(self, variable, key, domain='DATABASE'):
        """Set variable. Priority: input, environment, config"""
        result = None
        if variable is not None:
            result = variable
        elif environ.get(key):
            result = environ.get(key)
        elif domain in self.config and key in self.config[domain]:
            result = path.expanduser(self.config[domain][key])
        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 todays 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_tag_today(self, tag):
        """Get today tasks with specific tag"""
        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
                TAGS.tags=(SELECT uuid FROM {self.TABLE_TAG}
                             WHERE title='{tag}') 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_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,
                    (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
                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 COLLATE NOCASE
                """
        return self.execute_query(query)

    def get_areas(self):
        """Get areas."""
        query = f"""
                SELECT
                    AREA.uuid AS uuid,
                    AREA.title AS title,
                    (SELECT COUNT(uuid)
                        FROM TMTask AS PROJECT
                        WHERE
                        PROJECT.area = AREA.uuid AND
                        PROJECT.{self.IS_NOT_TRASHED} AND
                        PROJECT.{self.IS_OPEN}
                    ) AS size
                FROM
                    {self.TABLE_AREA} AS AREA
                ORDER BY AREA.title COLLATE NOCASE
                """
        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 COLLATE NOCASE DESC
            """
        return self.execute_query(query)

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

AlexanderWillner / KanbanView

Push — master ( db01a9...08f366 )

things3.things3.Things3.__init__() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity Many Parameters

Long Method

Complexity

Many Parameters

Duplication Side-by-Side

Filter issues like

things3.things3.Things3.init() C