Source code for tastic.workspace.workspace

#!/usr/local/bin/python
# encoding: utf-8
"""
*sorts the contents of all taskpaper files via workflow tags*

:Author:
    David Young

:Date Created:
    November  5, 2016
"""
################# GLOBAL IMPORTS ####################
import sys
import os
os.environ['TERM'] = 'vt100'
from fundamentals import tools
from fundamentals.renderer import list_of_dictionaries
from operator import itemgetter
import collections
import codecs
import textwrap
from tastic.tastic import document


[docs]class workspace(): """ *tools for sorting, archiving and indexing tasks and maintaining the contents of all taskpaper files within a given workspace* **Key Arguments:** - ``log`` -- logger - ``fileOrWorkspacePath`` -- the root path of the workspace you wish to sort the taskpaper docs within, or the path to a single taskpaper file - ``settings`` -- the settings dictionary **Usage:** To setup your logger, settings and database connections, please use the ``fundamentals`` package (`see tutorial here <http://fundamentals.readthedocs.io/en/latest/#tutorial>`_). To initiate a taskpaper workspace object, use the following: .. code-block:: python from tastic.workspace import workspace ws = workspace( log=log, settings=settings, fileOrWorkspacePath="/path/to/root/of/workspace" ) or to target a single taskpaper document use instead the path to the file: .. code-block:: python from tastic.workspace import workspace ws = workspace( log=log, settings=settings, fileOrWorkspacePath="/path/to/doc.taskpaper" ) """ # Initialisation
[docs] def __init__( self, log, fileOrWorkspacePath, settings=False ): self.log = log log.debug("instansiating a new 'sort' object") self.settings = settings self.taskpaperPath = False self.workspaceRoot = False # xt-self-arg-tmpx # INITIAL ACTIONS # ARE WE DEALING WITH A WORKSPACE DIRECTORY OR SINGLE FILE if os.path.isfile(fileOrWorkspacePath): self.taskpaperPath = fileOrWorkspacePath else: self.workspaceRoot = fileOrWorkspacePath self.taskpaperFiles = self._get_all_taskpaper_files() return None
[docs] def sort(self): """ *sort the workspace or individual taskpaper document via the workflow tags found in the settings file* **Usage:** To sort all of the taskpaper documents in the workspace via the workflow tag set with the settings file, for example: .. code-block:: yaml workflowTags: "@due, @flag, @hold, @next, @someday, @wait" use the ``sort()`` method: .. code-block:: python ws.sort() """ self.log.info('starting the ``sort`` method') for f in self.taskpaperFiles: self._sort_tp_file(f) self.log.info('completed the ``sort`` method') return None
[docs] def archive_done( self): """*move done tasks from the document's 'Archive' project into an adjacent markdown tasklog file* **Usage:** To move the archived tasks within a workspace's taskpaper docs into ``-tasklog.md`` files use the ``archive_done()`` method: .. code-block:: python ws.archive_done() """ self.log.info('starting the ``archive_done`` method') for f in self.taskpaperFiles: self._archive_tp_file_done_tasks(f) self.log.info('completed the ``archive_done`` method') return None
def _get_all_taskpaper_files( self): """*get a list of all the taskpaper filepaths in the workspace* **Return:** - ``taskpaperFiles`` -- a list of paths to all the taskpaper files within the workspace """ self.log.info('starting the ``_get_all_taskpaper_files`` method') if self.workspaceRoot: from fundamentals.files import recursive_directory_listing theseFiles = recursive_directory_listing( log=self.log, baseFolderPath=self.workspaceRoot, whatToList="files" # all | files | dirs ) taskpaperFiles = [] taskpaperFiles[:] = [f for f in theseFiles if os.path.splitext(f)[ 1] == ".taskpaper"] else: taskpaperFiles = [self.taskpaperPath] self.log.info('completed the ``_get_all_taskpaper_files`` method') return taskpaperFiles def _sort_tp_file( self, taskpaperPath): """*sort individual taskpaper documents* **Key Arguments:** - ``taskpaperPath`` -- path to a taskpaper file **Return:** - None """ self.log.info('starting the ``_sort_tp_file`` method') # OPEN TASKPAPER FILE self.log.info("sorting taskpaper file %(taskpaperPath)s" % locals()) doc = document(taskpaperPath) doc.tidy() doc.sort_tasks(self.settings["workflowTags"]) doc.sort_projects(self.settings["workflowTags"]) doc.save() self.log.info('completed the ``_sort_tp_file`` method') return None def _archive_tp_file_done_tasks( self, taskpaperPath): """* archive tp file done tasks* **Key Arguments:** - ``taskpaperPath`` -- path to a taskpaper file **Return:** - None """ self.log.info('starting the ``_archive_tp_file_done_tasks`` method') self.log.info("archiving taskpaper file %(taskpaperPath)s" % locals()) taskLog = {} mdArchiveFile = taskpaperPath.replace(".taskpaper", "-tasklog.md") exists = os.path.exists(mdArchiveFile) if exists: pathToReadFile = mdArchiveFile try: self.log.debug("attempting to open the file %s" % (pathToReadFile,)) readFile = codecs.open( pathToReadFile, encoding='utf-8', mode='r') thisData = readFile.read() readFile.close() except IOError, e: message = 'could not open the file %s' % (pathToReadFile,) self.log.critical(message) raise IOError(message) readFile.close() table = False for l in thisData.split("\n"): l = l.encode("utf-8") if ":---" in l: table = True continue if table == True and len(l) and l[0] == "|": dictt = collections.OrderedDict(sorted({}.items())) columns = l.split("|") dictt["task"] = columns[1].strip().decode("utf-8") dictt["completed"] = columns[2].strip().decode("utf-8") dictt["project"] = columns[3].strip().decode("utf-8") taskLog[dictt["task"] + dictt["completed"] + dictt["project"]] = dictt doc = document(taskpaperPath) aProject = doc.get_project("Archive") if not aProject: return doneTasks = aProject.tagged_tasks("@done") for task in doneTasks: dateCompleted = "" project = "" for t in task.tags: if "done" in t: dateCompleted = t.replace("done", "").replace( "(", "").replace(")", "") if "project(" in t: project = t.replace("project", "").replace( "(", "").replace(")", "") dictt = collections.OrderedDict(sorted({}.items())) notes = "" if task.notes: for n in task.notes: if len(notes) and notes[-2:] != ". ": if notes[-1] == ".": notes += " " else: notes += ". " notes += n.title if len(notes): notes = "<br><br>**NOTES:**<br>" + \ "<br>".join(textwrap.wrap( notes, 120, break_long_words=True)) dictt["task"] = "<br>".join(textwrap.wrap(task.title[ 2:], 120, break_long_words=True)) + notes dictt["task"] = dictt["task"].encode("utf-8") dictt["completed"] = dateCompleted dictt["project"] = project # SET ENCODE ERROR RETURN VALUE # RECODE INTO ASCII dictt["task"] = dictt["task"].decode("utf-8") dictt["completed"] = dictt["completed"].decode("utf-8") dictt["project"] = dictt["project"].decode("utf-8") taskLog[dictt["task"] + dictt["completed"] + dictt["project"]] = dictt taskLog = taskLog.values() taskLog = sorted(taskLog, key=itemgetter('task'), reverse=True) taskLog = sorted(taskLog, key=itemgetter('project'), reverse=True) taskLog = sorted(taskLog, key=itemgetter('completed'), reverse=True) dataSet = list_of_dictionaries( log=self.log, listOfDictionaries=taskLog ) markdownData = dataSet.markdown(filepath=None) try: self.log.debug("attempting to open the file %s" % (mdArchiveFile,)) writeFile = codecs.open(mdArchiveFile, encoding='utf-8', mode='w') except IOError, e: message = 'could not open the file %s' % (mdArchiveFile,) self.log.critical(message) raise IOError(message) writeFile.write(markdownData.decode("utf-8")) writeFile.close() aProject.delete() doc.save() self.log.info('completed the ``_archive_tp_file_done_tasks`` method') return None
# use the tab-trigger below for new method # xt-class-method