things_cli.cli.ThingsCLI.csv_converter()   A
last analyzed

Complexity

Conditions 3

Size

Total Lines 10
Code Lines 9

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 3
eloc 9
nop 3
dl 0
loc 10
rs 9.95
c 0
b 0
f 0
1
#!/usr/bin/env python3
2
3
"""A simple Python 3 CLI to read your Things app data."""
4
5
from __future__ import print_function
6
7
import argparse
8
import csv
9
from datetime import datetime
10
from io import StringIO
11
import json
12
import sys
13
from typing import Dict
14
import webbrowser
15
from xml.dom import minidom
16
import xml.etree.ElementTree as ETree
17
from xml.etree.ElementTree import Element, SubElement
18
19
import argcomplete  # type: ignore
0 ignored issues
show
introduced by
Unable to import 'argcomplete'
Loading history...
20
import things as api
0 ignored issues
show
introduced by
Unable to import 'things'
Loading history...
21
22
from things_cli import __version__
23
24
25
class ThingsCLI:  # pylint: disable=too-many-instance-attributes
26
    """A simple Python 3 CLI to read your Things app data."""
27
28
    print_json = False
29
    print_csv = False
30
    print_gantt = False
31
    print_opml = False
32
    # anonymize = False
33
    database = None
34
    recursive = False
35
    filter_project = None
36
    filter_area = None
37
    filter_tag = None
38
    only_projects = None
39
40
    def __init__(self, database=None):
41
        """Initialize class."""
42
        self.database = database
43
44
    def print_tasks(self, tasks):
45
        """Print a task."""
46
47
        if self.only_projects:
48
            for task in tasks:
49
                task["items"] = (
50
                    [
51
                        items
52
                        for items in task["items"]
53
                        if items["type"] in ["area", "project"]
54
                    ]
55
                    if task.get("items")
56
                    else []
57
                )
58
                for items in task["items"]:
59
                    items["items"] = (
60
                        [
61
                            sub_items
62
                            for sub_items in items["items"]
63
                            if sub_items["type"] in ["area", "project"]
64
                        ]
65
                        if items.get("items")
66
                        else []
67
                    )
68
69
        if self.print_json:
70
            print(json.dumps(tasks))
71
        elif self.print_opml:
72
            print(self.opml_dumps(tasks))
73
        elif self.print_csv:
74
            print(self.csv_dumps(tasks))
75
        elif self.print_gantt:
76
            print("gantt")
77
            print("  dateFormat  YYYY-MM-DD")
78
            print("  title       Things To-Dos")
79
            print("  excludes    weekends")
80
            print(self.gantt_dumps(tasks))
81
        else:
82
            print(self.txt_dumps(tasks), end="")
83
84
    def gantt_dumps(self, tasks, array=None):
85
        """Convert tasks into mermaid-js GANTT."""
86
87
        result = ""
88
89
        if array is None:
90
            array = {}
91
92
        for task in tasks:
93
            ThingsCLI.gantt_add_task(task, array)
94
            self.gantt_dumps(task.get("items", []), array)
95
96
        for group in array:
97
            result += f"  section {group}\n"
98
            for item in array[group]:
99
                result += item
100
101
        return result
102
103
    @staticmethod
104
    def gantt_add_task(task, array):
105
        """Add a task to a mermaid-js GANTT."""
106
107
        context = (
108
            task.get("project_title", None)
109
            or task.get("area_title", None)
110
            or task.get("heading_title", None)
111
            or task.get("start", None)
112
            or ""
113
        )
114
115
        title = task["title"].replace(":", " ")
116
        start = task.get("start_date")
117
        deadline = task.get("deadline") or "1h"
118
        if not start and deadline != "1h":
119
            start = deadline
120
        if start == deadline:
121
            deadline = "1h"
122
        if deadline == "1h":
123
            visual = ":milestone"
124
        else:
125
            visual = ":active"
126
            # lint-ignore todo: if in the past: done
127
        if start and not task.get("stop_date"):
128
            if context not in array:
129
                array[context] = []
130
            if not "".join(s for s in array[context] if title.lower() in s.lower()):
131
                array[context].append(f"    {title} {visual}, {start}, {deadline}\n")
132
133
    def csv_dumps(self, tasks):
134
        """Convert tasks into CSV."""
135
136
        fieldnames = []
137
        self.csv_header(tasks, fieldnames)
138
        if "items" in fieldnames:
139
            fieldnames.remove("items")
140
        if "checklist" in fieldnames:
141
            fieldnames.remove("checklist")
142
        fieldnames.sort()  # Because task content affects the order that fields are discovered above
143
144
        output = StringIO()
145
        writer = csv.DictWriter(
146
            output, fieldnames=fieldnames, delimiter=";", escapechar="\\"
147
        )
148
        writer.writeheader()
149
150
        self.csv_converter(tasks, writer)
151
152
        return output.getvalue()
153
154
    def csv_header(self, tasks, fieldnames):
155
        """Convert tasks into CSV header."""
156
        for task in tasks:
157
            fieldnames.extend(field for field in task if field not in fieldnames)
158
            self.csv_header(task.get("items", []), fieldnames)
159
160
    def csv_converter(self, tasks, writer):
161
        """Convert tasks into CSV."""
162
        if tasks is True:
163
            return
164
        for task in tasks:
165
            self.csv_converter(task.get("items", []), writer)
166
            task.pop("items", [])
167
            self.csv_converter(task.get("checklist", []), writer)
168
            task.pop("checklist", [])
169
            writer.writerow(task)
170
171
    def opml_dumps(self, tasks):
172
        """Convert tasks into OPML."""
173
174
        top = Element("opml")
175
        head = SubElement(top, "head")
176
        SubElement(head, "title").text = "Things 3 Database"
177
        body = SubElement(top, "body")
178
179
        self.opml_convert(tasks, body)
180
181
        return minidom.parseString(ETree.tostring(top)).toprettyxml(indent="   ")
182
183
    def opml_convert(self, tasks, top):
184
        """Print pretty OPML of selected tasks."""
185
186
        if tasks is True:
187
            return
188
        for task in tasks:
189
            area = SubElement(top, "outline")
190
            text = task["title"]
191
            if task.get("start_date"):
192
                text = f"{text} (Scheduled: {task['start_date']})"
193
            elif task.get("start"):
194
                text = f"{text} ({task['start']})"
195
            area.set("text", text)
196
            self.opml_convert(task.get("items", []), area)
197
            task.pop("items", [])
198
            self.opml_convert(task.get("checklist", []), area)
199
            task.pop("checklist", [])
200
201
    def txt_dumps(self, tasks, indentation="", result=""):
202
        """Print pretty text version of selected tasks."""
203
204
        if tasks is True:
205
            return result
206
        for task in tasks:
207
            title = task["title"]
208
            context = (
209
                task.get("project_title", None)
210
                or task.get("area_title", None)
211
                or task.get("heading_title", None)
212
                or task.get("start", None)
213
            )
214
            start = task.get("start_date", None)
215
            details = " | ".join(filter(None, [start, context]))
216
            result = result + f"{indentation}- {title} ({details})\n"
217
            result = self.txt_dumps(task.get("items", []), indentation + "  ", result)
218
            task.pop("items", [])
219
            result = self.txt_dumps(
220
                task.get("checklist", []), indentation + "  ", result
221
            )
222
223
        return result
224
225
    @classmethod
226
    def print_unimplemented(cls, command):
227
        """Show warning that method is not yet implemented."""
228
        print(f"command '{command}' not implemented yet", file=sys.stderr)
229
230
    @classmethod
231
    def get_parser(cls):
232
        """Create command line argument parser."""
233
        parser = argparse.ArgumentParser(description="Simple read-only Thing 3 CLI.")
234
235
        subparsers = parser.add_subparsers(
236
            help="", metavar="command", required=True, dest="command"
237
        )
238
239
        ################################
240
        # Core database methods
241
        ################################
242
        subparsers.add_parser("inbox", help="Shows inbox tasks")
243
        subparsers.add_parser("today", help="Shows todays tasks")
244
        subparsers.add_parser("upcoming", help="Shows upcoming tasks")
245
        subparsers.add_parser("anytime", help="Shows anytime tasks")
246
        subparsers.add_parser("completed", help="Shows completed tasks")
247
        subparsers.add_parser("someday", help="Shows someday tasks")
248
        subparsers.add_parser("canceled", help="Shows canceled tasks")
249
        subparsers.add_parser("trash", help="Shows trashed tasks")
250
        subparsers.add_parser("todos", help="Shows all todos")
251
        subparsers.add_parser("all", help="Shows all tasks")
252
        subparsers.add_parser("areas", help="Shows all areas")
253
        subparsers.add_parser("projects", help="Shows all projects")
254
        subparsers.add_parser("logbook", help="Shows completed tasks")
255
        subparsers.add_parser("logtoday", help="Shows tasks completed today")
256
        subparsers.add_parser("createdtoday", help="Shows tasks created today")
257
        subparsers.add_parser("tags", help="Shows all tags ordered by their usage")
258
        subparsers.add_parser("deadlines", help="Shows tasks with due dates")
259
260
        ################################
261
        # Additional functions
262
        ################################
263
        subparsers.add_parser("feedback", help="Give feedback")
264
        subparsers.add_parser(
265
            "search", help="Searches for a specific task"
266
        ).add_argument("string", help="String to search for")
267
268
        ################################
269
        # To be implemented in things.py
270
        ################################
271
        # subparsers.add_parser("repeating", help="Shows all repeating tasks")
272
        # subparsers.add_parser("subtasks", help="Shows all subtasks")
273
        # subparsers.add_parser("headings", help="Shows headings")
274
275
        ################################
276
        # To be converted from https://github.com/alexanderwillner/things.sh
277
        ################################
278
        # subparsers.add_parser("backlog", help="Shows backlog tasks")
279
        # subparsers.add_parser("empty", help="Shows projects that are empty")
280
        # subparsers.add_parser("hours", help="Shows hours planned today")
281
        # subparsers.add_parser("ical", help="Shows tasks ordered by due date as iCal")
282
        # subparsers.add_parser("lint", help="Shows tasks that float around")
283
        # subparsers.add_parser(
284
        #     "mostClosed", help="Shows days when most tasks were closed"
285
        # )
286
        # subparsers.add_parser(
287
        #     "mostCancelled", help="Shows days when most tasks were cancelled"
288
        # )
289
        # subparsers.add_parser(
290
        #     "mostTrashed", help="Shows days when most tasks were trashed"
291
        # )
292
        # subparsers.add_parser(
293
        #     "mostCreated", help="Shows days when most tasks were created"
294
        # )
295
        # subparsers.add_parser("mostTasks", help="Shows projects that have most tasks")
296
        # subparsers.add_parser(
297
        #     "mostCharacters", help="Shows tasks that have most characters"
298
        # )
299
        # subparsers.add_parser("nextish", help="Shows all nextish tasks")
300
        # subparsers.add_parser("old", help="Shows all old tasks")
301
        # subparsers.add_parser("schedule", help="Schedules an event using a template")
302
        # subparsers.add_parser("stat", help="Provides a number of statistics")
303
        # subparsers.add_parser("statcsv", help="Exports some statistics as CSV")
304
        # subparsers.add_parser("tag", help="Shows all tasks with the waiting for tag")
305
        # subparsers.add_parser(
306
        #     "waiting", help="Shows all tasks with the waiting for tag"
307
        # )
308
309
        ################################
310
        # To be converted from https://github.com/alexanderwillner/things.sh
311
        ################################
312
        # parser.add_argument("-a", "--anonymize",
313
        #                     action="store_true", default=False,
314
        #                     help="anonymize output", dest="anonymize")
315
316
        parser.add_argument(
317
            "-p", "--filter-project", dest="filter_project", help="filter by project (UUID)"
318
        )
319
        parser.add_argument(
320
            "-a", "--filter-area", dest="filter_area", help="filter by area (UUID)"
321
        )
322
        parser.add_argument(
323
            "-t", "--filtertag", dest="filter_tag", help="filter by tag"
324
        )
325
        parser.add_argument(
326
            "-e",
327
            "--only-projects",
328
            action="store_true",
329
            default=False,
330
            dest="only_projects",
331
            help="export only projects",
332
        )
333
        parser.add_argument(
334
            "-o",
335
            "--opml",
336
            action="store_true",
337
            default=False,
338
            help="output as OPML",
339
            dest="opml",
340
        )
341
342
        parser.add_argument(
343
            "-j",
344
            "--json",
345
            action="store_true",
346
            default=False,
347
            help="output as JSON",
348
            dest="json",
349
        )
350
351
        parser.add_argument(
352
            "-c",
353
            "--csv",
354
            action="store_true",
355
            default=False,
356
            help="output as CSV",
357
            dest="csv",
358
        )
359
360
        parser.add_argument(
361
            "-g",
362
            "--gantt",
363
            action="store_true",
364
            default=False,
365
            help="output as mermaid-js GANTT",
366
            dest="gantt",
367
        )
368
369
        parser.add_argument(
370
            "-r",
371
            "--recursive",
372
            help="in-depth output",
373
            dest="recursive",
374
            default=False,
375
            action="store_true",
376
        )
377
378
        parser.add_argument(
379
            "-d", "--database", help="set path to database", dest="database"
380
        )
381
382
        parser.add_argument(
383
            "--version",
384
            "-v",
385
            action="version",
386
            version=f"%(prog)s (version {__version__})",
387
        )
388
389
        argcomplete.autocomplete(parser)
390
391
        return parser
392
393
    def defaults(self):
394
        """Set default options for the new API."""
395
        return {
396
            "project": self.filter_project,
397
            "area": self.filter_area,
398
            "tag": self.filter_tag,
399
            "include_items": self.recursive,
400
            "filepath": self.database,
401
        }
402
403
    def main(self, args=None):
404
        """Start the main app."""
405
406
        if args is None:
407
            self.main(ThingsCLI.get_parser().parse_args())
408
        else:
409
            self.print_json = args.json
410
            self.print_csv = args.csv
411
            self.print_gantt = args.gantt
412
            self.print_opml = args.opml
413
            self.database = args.database or self.database
414
            self.filter_project = args.filter_project or None
415
            self.filter_area = args.filter_area or None
416
            self.filter_tag = args.filter_tag or None
417
            self.only_projects = args.only_projects or None
418
            self.recursive = args.recursive
419
            # self.anonymize = args.anonymize
420
            # self.things3.anonymize = self.anonymize ## not implemented
421
            defaults = self.defaults()
422
423
            self.parse_command(defaults, args)
424
425
    def parse_command(self, defaults: Dict, args):
426
        """Handle given command."""
427
428
        command = args.command
429
430
        if command == "tags":
431
            defaults.pop("tag")
432
            defaults.pop("project")
433
        if command in ["all", "areas"]:
434
            defaults.pop("area")
435
            defaults.pop("project")
436
437
        if command == "all":
438
            inbox = api.inbox(**defaults)
439
            today = api.today(**defaults)
440
            upcoming = api.upcoming(**defaults)
441
            anytime = api.anytime(**defaults)
442
            someday = api.someday(**defaults)
443
            logbook = api.logbook(**defaults)
444
445
            no_area = api.projects(**defaults)
446
            areas = api.areas(**defaults)
447
            structure = [
448
                {"title": "Inbox", "items": inbox},
449
                {"title": "Today", "items": today},
450
                {"title": "Upcoming", "items": upcoming},
451
                {"title": "Anytime", "items": anytime},
452
                {"title": "Someday", "items": someday},
453
                {"title": "Logbook", "items": logbook},
454
                {"title": "No Area", "items": no_area},
455
                {"title": "Areas", "items": areas},
456
            ]
457
            self.print_tasks(structure)
458
        elif command == "logtoday":
459
            today = datetime.now().strftime("%Y-%m-%d")
460
            result = getattr(api, "logbook")(**defaults, stop_date=today)
461
            self.print_tasks(result)
462
        elif command == "createdtoday":
463
            result = getattr(api, "last")("1d")
464
            self.print_tasks(result)
465
        elif command == "upcoming":
466
            result = getattr(api, command)(**defaults)
467
            result.sort(key=lambda task: task["start_date"], reverse=False)
468
            self.print_tasks(result)
469
        elif command == "search":
470
            self.print_tasks(
471
                api.search(
472
                    args.string,
473
                    filepath=self.database,
474
                    include_items=self.recursive,
475
                )
476
            )
477
        elif command == "feedback":  # pragma: no cover
478
            webbrowser.open("https://github.com/thingsapi/things-cli/issues")
479
        elif command in dir(api):
480
            self.print_tasks(getattr(api, command)(**defaults))
481
        else:  # pragma: no cover
482
            ThingsCLI.print_unimplemented(command)
483
            sys.exit(3)
484
485
486
def main():
487
    """Start for CLI installation."""
488
    ThingsCLI().main()
489
490
491
if __name__ == "__main__":
492
    main()
493