File._taskModified() - Code Metrics - Inspection of "move modified code out of main_controller" - QTodoTxt/QTodoTxt2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — qml ( e78c5d...75cc72 )

by Olivier

created 2017-05-16 16:38 UTC

File._taskModified() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
c	0
b	0
f	0
dl	0
loc	7
rs	9.4285

import logging
import os

from PyQt5 import QtCore

from qtodotxt.lib.filters import DueTodayFilter, DueTomorrowFilter, DueThisWeekFilter, DueThisMonthFilter, DueOverdueFilter
from qtodotxt.lib.tasklib import Task

logger = logging.getLogger(__name__)


class File(QtCore.QObject):

    fileExternallyModified = QtCore.pyqtSignal()
    fileModified = QtCore.pyqtSignal(bool)

    def __init__(self):
        QtCore.QObject.__init__(self)
        self.newline = '\n'
        self.tasks = []
        self.filename = ''
        self._fileObserver = FileObserver()
        self._fileObserver.fileChangetSig.connect(self.fileExternallyModified)
        self.modified = False

    def __str__(self):
        return "File(filename:{}, tasks:{})".format(self.filename, self.tasks)

    __repr__ = __str__

    def load(self, filename):
        self._fileObserver.clear()
        with open(filename, 'rt', encoding='utf-8') as fd:
            lines = fd.readlines()
        self.filename = filename
        self._createTasksFromLines(lines)
        self._fileObserver.addPath(self.filename)

    def _createTasksFromLines(self, lines):
        self.tasks = []
        for line in lines:
            task_text = line.strip()
            if task_text:
                task = Task(task_text)
                self.tasks.append(task)
                task.modified.connect(self._taskModified)

    def _taskModified(self, task):
        print("FILE task modified", task)
        self.setModified(True)
        #if task not in self.tasks:
            #self.tasks.append(task)
        if not task.text:
            self.deleteTask(task)

    def setModified(self, val):
        self.modified = val
        self.fileModified.emit(val)

    def deleteTask(self, task):
        print("FILE delkete", task)
        self.tasks.remove(task)
        self.setModified(True)

    def addTask(self, task):
        self.tasks.append(task)
        task.modified.connect(self._taskModified)
        self.setModified(True)

    def connectTask(self, task):
        task.modified.connect(self._taskModified)

    def save(self, filename=''):
        logger.debug('File.save called with filename="%s"', filename)
        self._fileObserver.clear()
        if not filename and not self.filename:
            self.filename = self._createNewFilename()
        elif filename:
            self.filename = filename
        self._saveTasks()
        self.modified = False
        self.fileModified.emit(False)
        self._fileObserver.addPath(self.filename)

    @staticmethod
    def _createNewFilename():
        newFileName = os.path.expanduser('~/todo.txt')
        if not os.path.isfile(newFileName):
            return newFileName
        for counter in range(0, 100):
            newFileName = os.path.expanduser('~/todo.{}.txt.'.format(counter))
            if not os.path.isfile(newFileName):
                return newFileName
        return os.path.expanduser('~/todo.0.txt')

    def _saveTasks(self):
        with open(self.filename, 'wt', encoding='utf-8') as fd:
            fd.writelines([(task.text + self.newline) for task in self.tasks])
        logger.debug('%s was saved to disk.', self.filename)

    def saveDoneTask(self, task):
        doneFilename = os.path.join(os.path.dirname(self.filename), 'done.txt')
        with open(doneFilename, 'at', encoding='utf-8') as fd:
            fd.write(task.text + self.newline)
        logger.debug('"%s" was appended to "%s"', task.text, doneFilename)

    def getAllContexts(self):
        return self._getAllX("contexts")

    def getAllProjects(self):
        return self._getAllX("projects")

    def _getAllX(self, name):
        res = {}
        for task in self.tasks:
            for element in getattr(task, name):
                if element not in res:
                    res[element] = [0, 0]
                idx = 1 if task.is_complete else 0
                res[element][idx] += 1
        return res

    def getAllPriorities(self):
        return self._getAllX("priority")

    def getAllDueRanges(self):
        dueRanges = dict()
        filters = [DueTodayFilter(), DueTomorrowFilter(), DueThisWeekFilter(), DueThisMonthFilter(), DueOverdueFilter()]
        for task in self.tasks:
            idx = 1 if task.is_complete else 0
            for flt in filters:
                if flt.isMatch(task):
                    if not (flt in dueRanges):
                        dueRanges[flt] = [0, 0]
                    dueRanges[flt][idx] += 1
        return dueRanges

    def getTasksCounters(self):
        counters = dict({
            'All': [0, 0],
            'Uncategorized': [0, 0],
            'Contexts': [0, 0],
            'Projects': [0, 0],
            'Complete': [0, 0],
            'Priority': [0, 0],
            'Due': [0, 0]
        })
        for task in self.tasks:
            nbProjects = len(task.projects)
            nbContexts = len(task.contexts)
            idx = 1 if task.is_complete else 0
            counters['All'][idx] += 1
            if nbProjects > 0:
                counters['Projects'][idx] += 1
            if nbContexts > 0:
                counters['Contexts'][idx] += 1
            if nbContexts == 0 and nbProjects == 0:
                counters['Uncategorized'][idx] += 1
            if task.due:
                counters['Due'][idx] += 1
            if task.priority != "":
                counters['Priority'][idx] += 1
        return counters


class FileObserver(QtCore.QFileSystemWatcher):

    fileChangetSig = QtCore.pyqtSignal(str)
    dirChangetSig = QtCore.pyqtSignal(str)

    def __init__(self):
        logger.debug('Setting up FileObserver instance.')
        super().__init__()
        self.fileChanged.connect(self.fileChangedHandler)
        self.directoryChanged.connect(self.dirChangedHandler)

    def fileChangedHandler(self, path):
        logger.debug('Detected external file change for file %s', path)
        self.removePath(path)
        self.fileChangetSig.emit(path)

    def dirChangedHandler(self, path):
        logger.debug('Detected directory change for file %s', path)
        self.dirChangetSig.emit(path)

    def clear(self):
        if self.files():
            logger.debug('Clearing watchlist.')
            self.removePaths(self.files())


1			import logging
2			import os
3
4			from PyQt5 import QtCore
5
6			from qtodotxt.lib.filters import DueTodayFilter, DueTomorrowFilter, DueThisWeekFilter, DueThisMonthFilter, DueOverdueFilter
7			from qtodotxt.lib.tasklib import Task
8
9			logger = logging.getLogger(__name__)
10
11
12			class File(QtCore.QObject):
13
14			fileExternallyModified = QtCore.pyqtSignal()
15			fileModified = QtCore.pyqtSignal(bool)
16
17			def __init__(self):
18			QtCore.QObject.__init__(self)
19			self.newline = '\n'
20			self.tasks = []
21			self.filename = ''
22			self._fileObserver = FileObserver()
23			self._fileObserver.fileChangetSig.connect(self.fileExternallyModified)
24			self.modified = False
25
26			def __str__(self):
27			return "File(filename:{}, tasks:{})".format(self.filename, self.tasks)
28
29			__repr__ = __str__
30
31			def load(self, filename):
32			self._fileObserver.clear()
33			with open(filename, 'rt', encoding='utf-8') as fd:
34			lines = fd.readlines()
35			self.filename = filename
36			self._createTasksFromLines(lines)
37			self._fileObserver.addPath(self.filename)
38
39			def _createTasksFromLines(self, lines):
40			self.tasks = []
41			for line in lines:
42			task_text = line.strip()
43			if task_text:
44			task = Task(task_text)
45			self.tasks.append(task)
46			task.modified.connect(self._taskModified)
47
48			def _taskModified(self, task):
49			print("FILE task modified", task)
50			self.setModified(True)
51			#if task not in self.tasks:
52			#self.tasks.append(task)
53			if not task.text:
54			self.deleteTask(task)
55
56			def setModified(self, val):
57			self.modified = val
58			self.fileModified.emit(val)
59
60			def deleteTask(self, task):
61			print("FILE delkete", task)
62			self.tasks.remove(task)
63			self.setModified(True)
64
65			def addTask(self, task):
66			self.tasks.append(task)
67			task.modified.connect(self._taskModified)
68			self.setModified(True)
69
70			def connectTask(self, task):
71			task.modified.connect(self._taskModified)
72
73			def save(self, filename=''):
74			logger.debug('File.save called with filename="%s"', filename)
75			self._fileObserver.clear()
76			if not filename and not self.filename:
77			self.filename = self._createNewFilename()
78			elif filename:
79			self.filename = filename
80			self._saveTasks()
81			self.modified = False
82			self.fileModified.emit(False)
83			self._fileObserver.addPath(self.filename)
84
85			@staticmethod
86			def _createNewFilename():
87			newFileName = os.path.expanduser('~/todo.txt')
88			if not os.path.isfile(newFileName):
89			return newFileName
90			for counter in range(0, 100):
91			newFileName = os.path.expanduser('~/todo.{}.txt.'.format(counter))
92			if not os.path.isfile(newFileName):
93			return newFileName
94			return os.path.expanduser('~/todo.0.txt')
95
96			def _saveTasks(self):
97			with open(self.filename, 'wt', encoding='utf-8') as fd:
98			fd.writelines([(task.text + self.newline) for task in self.tasks])
99			logger.debug('%s was saved to disk.', self.filename)
100
101			def saveDoneTask(self, task):
102			doneFilename = os.path.join(os.path.dirname(self.filename), 'done.txt')
103			with open(doneFilename, 'at', encoding='utf-8') as fd:
104			fd.write(task.text + self.newline)
105			logger.debug('"%s" was appended to "%s"', task.text, doneFilename)
106
107			def getAllContexts(self):
108			return self._getAllX("contexts")
109
110			def getAllProjects(self):
111			return self._getAllX("projects")
112
113			def _getAllX(self, name):
114			res = {}
115			for task in self.tasks:
116			for element in getattr(task, name):
117			if element not in res:
118			res[element] = [0, 0]
119			idx = 1 if task.is_complete else 0
120			res[element][idx] += 1
121			return res
122
123			def getAllPriorities(self):
124			return self._getAllX("priority")
125
126			def getAllDueRanges(self):
127			dueRanges = dict()
128			filters = [DueTodayFilter(), DueTomorrowFilter(), DueThisWeekFilter(), DueThisMonthFilter(), DueOverdueFilter()]
129			for task in self.tasks:
130			idx = 1 if task.is_complete else 0
131			for flt in filters:
132			if flt.isMatch(task):
133			if not (flt in dueRanges):
134			dueRanges[flt] = [0, 0]
135			dueRanges[flt][idx] += 1
136			return dueRanges
137
138			def getTasksCounters(self):
139			counters = dict({
140			'All': [0, 0],
141			'Uncategorized': [0, 0],
142			'Contexts': [0, 0],
143			'Projects': [0, 0],
144			'Complete': [0, 0],
145			'Priority': [0, 0],
146			'Due': [0, 0]
147			})
148			for task in self.tasks:
149			nbProjects = len(task.projects)
150			nbContexts = len(task.contexts)
151			idx = 1 if task.is_complete else 0
152			counters['All'][idx] += 1
153			if nbProjects > 0:
154			counters['Projects'][idx] += 1
155			if nbContexts > 0:
156			counters['Contexts'][idx] += 1
157			if nbContexts == 0 and nbProjects == 0:
158			counters['Uncategorized'][idx] += 1
159			if task.due:
160			counters['Due'][idx] += 1
161			if task.priority != "":
162			counters['Priority'][idx] += 1
163			return counters
164
165
166			class FileObserver(QtCore.QFileSystemWatcher):
167
168			fileChangetSig = QtCore.pyqtSignal(str)
169			dirChangetSig = QtCore.pyqtSignal(str)
170
171			def __init__(self):
172			logger.debug('Setting up FileObserver instance.')
173			super().__init__()
174			self.fileChanged.connect(self.fileChangedHandler)
175			self.directoryChanged.connect(self.dirChangedHandler)
176
177			def fileChangedHandler(self, path):
178			logger.debug('Detected external file change for file %s', path)
179			self.removePath(path)
180			self.fileChangetSig.emit(path)
181
182			def dirChangedHandler(self, path):
183			logger.debug('Detected directory change for file %s', path)
184			self.dirChangetSig.emit(path)
185
186			def clear(self):
187			if self.files():
188			logger.debug('Clearing watchlist.')
189			self.removePaths(self.files())
190

QTodoTxt / QTodoTxt2

Push — qml ( e78c5d...75cc72 )

File._taskModified() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like