parser.py (7484B) - raw
1 #!/usr/bin/env python3 2 3 # Copyright (C) 2020 Oscar Benedito <oscar@oscarbenedito.com> 4 # 5 # This program is free software: you can redistribute it and/or modify 6 # it under the terms of the GNU Affero General Public License as 7 # published by the Free Software Foundation, either version 3 of the 8 # License, or (at your option) any later version. 9 # 10 # This program is distributed in the hope that it will be useful, 11 # but WITHOUT ANY WARRANTY; without even the implied warranty of 12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 13 # GNU Affero General Public License for more details. 14 # 15 # You should have received a copy of the GNU Affero General Public License 16 # along with this program. If not, see <https://www.gnu.org/licenses/>. 17 18 import sys 19 import re 20 import datetime 21 22 23 class Task: 24 def __init__(self, description, tags): 25 self.intervals = list() 26 self.description = description 27 self.tags = tags 28 29 def __repr__(self): 30 return 'Task({}, {}, {})'.format(self.intervals, self.description, 31 self.tags) 32 33 def id(self): 34 if len(self.intervals) == 0: 35 raise ValueError('Task has never been started. It needs a starting ' 36 'date to have an ID.') 37 return self.intervals[0][0].strftime('%Y%m%d%H%M%S') 38 39 def is_open(self): 40 return len(self.intervals) > 0 and self.intervals[-1][1] is None 41 42 def stop(self, time): 43 if not self.is_open(): 44 raise ValueError('Task is already stopped.') 45 46 self.intervals[-1][1] = time 47 48 def start(self, time): 49 if self.is_open(): 50 raise ValueError('Task is already started.') 51 52 self.intervals.append([time, None]) 53 54 55 class TimeTracker: 56 re_day = re.compile(r'^(\d\d\d\d)-(\d\d)-(\d\d):(?:[ \t]+#[ \t].*)?$') 57 re_new = re.compile(r'^(\d\d)(\d\d)(\d\d)?[ \t]+(.+?)(?:[ \t]+#[ \t].*)?$') 58 re_stop = re.compile(r'^(\d\d)(\d\d)(\d\d)?\.(?:[ \t]+#[ \t].*)?$') 59 re_rel = re.compile(r'^(\d\d)(\d\d)(\d\d)?(\^+)(?:[ \t]+#[ \t].*)?$') 60 re_abs = re.compile(r'^(\d\d)(\d\d)(\d\d)?\^(?:(\d\d\d\d)-(\d\d)-(\d\d)T)?' 61 '(\d\d)(\d\d)(\d\d)?(?:[ \t]+#[ \t].*)?$') 62 re_tag = re.compile(r'[ \t]#[a-zA-Z0-9_]+(?=[ \t])') 63 64 def __init__(self): 65 self.tasks = dict() 66 self.c_dt = None # current datetime 67 self.c_task = None # current task 68 self.l_task = None # last task 69 self.day_is_empty = None # true if there are tasks on the current day 70 71 def stop_task(self): 72 if self.c_task is not None and self.c_task.is_open(): 73 self.c_task.stop(self.c_dt) 74 return True 75 76 return False 77 78 def start_task(self, task): 79 if task != self.c_task: 80 self.l_task = self.c_task 81 self.c_task = task 82 83 self.c_task.start(self.c_dt) 84 85 def process_line(self, line): 86 line.strip() 87 if len(line) == 0 or line[:2] == '# ' or line[:2] == '#\t': 88 # whitespace/comment 89 return 90 91 if len(line) < 5 or (line[4].isdigit() and len(line) < 7): 92 raise ValueError('Incorrect syntax.') 93 94 if line[4] == '-': 95 # day change 96 self.process_line_day_change(line) 97 return 98 99 # depending on time precission (seconds or minutes) 100 i = 6 if line[4].isdigit() else 4 101 102 if line[i] == ' ' or line[i] == '\t': 103 # start new task 104 self.process_line_new_task(line) 105 return 106 107 if line[i] == '.': 108 # stop task 109 self.process_line_stop_task(line) 110 return 111 112 if line[i] == '^' and len(line) > i+1 and line[i+1].isdigit(): 113 # continue task by time stamp 114 self.proces_line_continue_time(line) 115 return 116 117 # continue last task 118 self.process_line_continue_last(line) 119 120 def process_datetime(self, hour, minute, second): 121 if self.c_dt is None: 122 raise ValueError('Task initialized before day declaration.') 123 124 hour = int(hour) 125 minute = int(minute) 126 second = int(second) if second is not None else 0 127 date = self.c_dt.replace(hour=hour, minute=minute, second=second) 128 129 if date < self.c_dt or (date == self.c_dt and not self.day_is_empty): 130 raise ValueError('Tasks are not in chronological order or have the ' 131 'same starting time.') 132 133 self.c_dt = date 134 self.day_is_empty = False 135 136 def process_line_day_change(self, line): 137 m = self.re_day.search(line) 138 if m is None: 139 raise ValueError('Incorrect syntax.') 140 141 date = datetime.datetime(int(m[1]), int(m[2]), int(m[3])) 142 143 if self.c_dt is not None and date <= self.c_dt: 144 raise ValueError('New date is older than last recorded time.') 145 146 self.c_dt = date 147 self.day_is_empty = True 148 149 def process_line_new_task(self, line): 150 m = self.re_new.search(line) 151 if m is None: 152 raise ValueError('Incorrect syntax.') 153 154 self.process_datetime(m[1], m[2], m[3]) 155 156 description = m[4] 157 tags = set() 158 for m_tag in self.re_tag.finditer(description): 159 tags.add(m_tag[2]) 160 description = self.re_tag.sub('', description).strip() 161 162 task = Task(description, tags) 163 164 self.stop_task() 165 self.start_task(task) 166 167 self.tasks[task.id()] = task 168 169 def process_line_stop_task(self, line): 170 m = self.re_stop.search(line) 171 if m is None: 172 raise ValueError('Incorrect syntax.') 173 174 self.process_datetime(m[1], m[2], m[3]) 175 176 if not self.stop_task(): 177 raise ValueError('No task to stop.') 178 179 def proces_line_continue_time(self, line): 180 m = self.re_abs.search(line) 181 if m is None: 182 raise ValueError('Incorrect syntax.') 183 184 self.process_datetime(m[1], m[2], m[3]) 185 186 iden = self.c_dt.strftime('%Y%m%d') if m[4] is None else m[4]+m[5]+m[6] 187 iden += m[7] + m[8] 188 iden += m[9] if m[9] is not None else '00' 189 190 n_task = self.tasks.get(iden) 191 if n_task is None: 192 pretty_date = '{}-{}-{} {}:{}:{}'.format(iden[:4], iden[4:6], 193 iden[6:8], iden[8:10], 194 iden[10:12], iden[12:14]) 195 raise ValueError('No task started on {}.'.format(pretty_date)) 196 197 self.stop_task() 198 self.start_task(n_task) 199 200 def process_line_continue_last(self, line): 201 m = self.re_rel.search(line) 202 if m is None: 203 raise ValueError('Incorrect syntax.') 204 205 self.process_datetime(m[1], m[2], m[3]) 206 207 n_task = self.l_task if self.stop_task() else self.c_task 208 if n_task is None: 209 raise ValueError('No task to resume.') 210 211 self.stop_task() 212 self.start_task(n_task) 213 214 215 def main(): 216 if len(sys.argv) != 2: 217 print('This program takes one argument (the file with the data).') 218 exit(1) 219 220 with open(sys.argv[1], 'r') as f: 221 lines = f.read().splitlines() 222 223 tt = TimeTracker() 224 for i, line in enumerate(lines): 225 try: 226 tt.process_line(line) 227 except ValueError as e: 228 print('Error on line {}: {}'.format(i+1, e)) 229 exit(1) 230 231 print(tt.tasks) 232 233 234 if __name__ == "__main__": 235 main()