plomlompom.com Git - misc/blob - new_todo/todo.py

   1 #!/usr/bin/env python3
   2 from sqlite3 import connect as sql_connect
   3 from http.server import BaseHTTPRequestHandler
   4 from urllib.parse import parse_qs
   5 from datetime import datetime, timedelta
   6
   7
   8
   9 PATH_DB_SCHEMA='init.sql'
  10 HTTP_PORT=8082
  11 HTML_DIR='html'
  12
  13
  14
  15 DATE_FORMAT = '%Y-%m-%d'
  16
  17
  18
  19 ######## basic system stuff ############
  20
  21 class HandledException(Exception):
  22
  23     def nice_exit(self):
  24         from sys import exit as sys_exit
  25         print(f'ABORTING: {self}')
  26         sys_exit(1)
  27
  28
  29
  30 class TodoDBFile:
  31
  32     def __init__(self, path):
  33         from os.path import isfile
  34         self.path = path
  35         if not isfile(self.path):
  36             self.make_new_if_wanted_else_abort()
  37         self.validate_schema()
  38
  39     def make_new_if_wanted_else_abort(self):
  40         create_question = f'Database file not found: {self.path}. Create? Y/n\n'
  41         msg_on_no = 'Interpreting reply as "no", but cannot run without database file.'
  42         legal_yesses = {'y', 'yes'}
  43         create_reply = input(create_question)
  44         if not create_reply.lower() in legal_yesses:
  45             raise HandledException(msg_on_no)
  46         with sql_connect(self.path) as conn:
  47             with open(PATH_DB_SCHEMA, 'r') as f:
  48                 conn.executescript(f.read())
  49
  50     def validate_schema(self):
  51         sql_for_schema = 'SELECT sql FROM sqlite_master ORDER BY sql'
  52         msg_wrong_schema = 'Database has wrong tables schema. Diff:\n'
  53         with sql_connect(self.path) as conn:
  54             schema = ';\n'.join([r[0] for r in conn.execute(sql_for_schema) if r[0]]) + ';'
  55             with open(PATH_DB_SCHEMA, 'r') as f:
  56                 stored_schema = f.read().rstrip()
  57                 if schema != stored_schema:
  58                     from difflib import Differ
  59                     d = Differ()
  60                     diff_msg = d.compare(schema.splitlines(), stored_schema.splitlines())
  61                     raise HandledException(msg_wrong_schema + '\n'.join(diff_msg))
  62
  63     def backup(self):
  64         from shutil import copy2
  65         from random import randint
  66         copy2(self.path, f'{self.path}.bak.0')
  67         for i in range(0, 6):
  68             if 1 == randint(0, 2**i):
  69                 copy2(self.path, f'{self.path}.bak.{i+1}')
  70                 break
  71
  72
  73
  74 class TodoDBConnection:
  75
  76     def __init__(self, db_file):
  77         self.file = db_file
  78         self.conn = sql_connect(self.file.path)
  79
  80     def commit(self):
  81         self.file.backup()
  82         self.conn.commit()
  83
  84     def exec(self, code, inputs=None):
  85         if not inputs:
  86             inputs = []
  87         return self.conn.execute(code, inputs)
  88
  89     def close(self):
  90         self.conn.close()
  91
  92
  93
  94 ######## models ############
  95
  96 class VersionedAttribute:
  97
  98     def __init__(self, db_conn, parent, name, default):
  99         self.parent = parent
 100         self.name = name
 101         self.default = default
 102         self.history = {}
 103         for row in db_conn.exec(f'SELECT * FROM {self.table_name} WHERE template = ?',
 104                                 (self.parent.id_,)):
 105             self.history[row[1]] = row[2]
 106
 107     @property
 108     def table_name(self):
 109         return f'versioned_{self.name}s'
 110
 111     def save(self, db_conn):
 112         for date, value in self.history.items():
 113             db_conn.exec(f'REPLACE INTO {self.table_name} VALUES (?, ?, ?)',
 114                          (self.parent.id_, date, value))
 115
 116     @property
 117     def newest_date(self):
 118         return sorted(self.history.keys())[-1]
 119
 120     @property
 121     def newest(self):
 122         if 0 == len(self.history):
 123             if self.parent.forked_from:
 124                 return getattr(self.parent.forked_from, self.name).newest
 125             return self.default
 126         return self.history[self.newest_date]
 127
 128     def set(self, value):
 129         if 0 == len(self.history) or value != self.history[self.newest_date]:
 130             self.history[Day.todays_date(with_time=True)] = value
 131
 132
 133
 134 class Day:
 135
 136     def __init__(self, date, comment=''):
 137         self.date = date
 138         self.datetime = self.__class__.date_valid(self.date)
 139         if not self.datetime:
 140             raise HandledException(f'Provided date is of wrong format: {date}')
 141         self.comment = comment
 142
 143     @classmethod
 144     def from_row(cls, row):
 145         return cls(row[0], row[1])
 146
 147     @classmethod
 148     def all(cls, db_conn, ensure_betweens=False, date_range=('', '')):
 149         legal_day_names = {'yesterday', 'today', 'tomorrow'}
 150         for val in [val for val in date_range
 151                     if val != '' and val not in legal_day_names
 152                     and not cls.date_valid(val)]:
 153             raise HandledException(f'Provided date is of wrong format: {val}')
 154         start_str = date_range[0] if date_range[0] else '2024-01-01'
 155         end_str = date_range[1] if date_range[1] else '2030-12-31'
 156         start_date = getattr(cls, f'{start_str}s_date')() if start_str in legal_day_names else start_str
 157         end_date = getattr(cls, f'{end_str}s_date')() if end_str in legal_day_names else end_str
 158         if start_date > end_date:
 159             temp = end_date
 160             end_date = start_date
 161             start_date = temp
 162         days = []
 163         for row in db_conn.exec('SELECT * FROM days WHERE date >= ? AND date <= ?',
 164                                 (start_date, end_date)):
 165             days += [cls.from_row(row)]
 166         if ensure_betweens:
 167             if '' != date_range[0] and not start_date in [d.date for d in days]:
 168                 days += [cls(start_date)]
 169             if '' != date_range[1] and not end_date in [d.date for d in days]:
 170                 days += [cls(end_date)]
 171         days.sort()
 172         if ensure_betweens and len(days) > 1:
 173             gapless_days = []
 174             for i, day in enumerate(days):
 175                 gapless_days += [day]
 176                 if i < len(days) - 1:
 177                     while day.next_date != days[i+1].date:
 178                         day = Day(day.next_date)
 179                         gapless_days += [day]
 180             days = gapless_days
 181         return days
 182
 183     @classmethod
 184     def by_date(cls, db_conn, date, make_if_none=False):
 185         for row in db_conn.exec('SELECT * FROM days WHERE date = ?', (date,)):
 186             return cls.from_row(row)
 187         return cls(date) if make_if_none else None
 188
 189     def save(self, db_conn):
 190         db_conn.exec('REPLACE INTO days VALUES (?, ?)', (self.date, self.comment))
 191
 192     @classmethod
 193     def date_valid(cls, date):
 194         try:
 195             result = datetime.strptime(date, DATE_FORMAT)
 196         except ValueError:
 197             return None
 198         return result
 199
 200     @classmethod
 201     def todays_date(cls, with_time=False):
 202         cut_length = 19 if with_time else 10
 203         return str(datetime.now())[:cut_length]
 204
 205     @classmethod
 206     def yesterdays_date(cls):
 207         return str(datetime.now() - timedelta(days=1))[:10]
 208
 209     @classmethod
 210     def tomorrows_date(cls):
 211         return str(datetime.now() + timedelta(days=1))[:10]
 212
 213     @property
 214     def next_date(self):
 215         next_datetime = self.datetime + timedelta(days=1)
 216         return next_datetime.strftime(DATE_FORMAT)
 217
 218     @property
 219     def prev_date(self):
 220         prev_datetime = self.datetime - timedelta(days=1)
 221         return prev_datetime.strftime(DATE_FORMAT)
 222
 223     @property
 224     def weekday(self):
 225         return self.datetime.strftime('%A')
 226
 227     def __eq__(self, other):
 228         return self.date == other.date
 229
 230     def __lt__(self, other):
 231         return self.date < other.date
 232
 233
 234
 235 class TodoTemplate:
 236
 237     def __init__(self, db_conn, id_):
 238         self.id_ = id_
 239         self.forked_from = None
 240         self.title = VersionedAttribute(db_conn, self, 'title', 'UNNAMED')
 241         self.default_effort = VersionedAttribute(db_conn, self, 'default_effort', 1.0)
 242         self.description = VersionedAttribute(db_conn, self, 'description', '')
 243
 244     @classmethod
 245     def from_row(cls, db_conn, row):
 246         return cls(db_conn, row[0])
 247
 248     @classmethod
 249     def all(cls, db_conn):
 250         tmpls = []
 251         for row in db_conn.exec('SELECT * FROM templates'):
 252             tmpls += [cls.from_row(db_conn, row)]
 253         return tmpls
 254
 255     @classmethod
 256     def by_id(cls, db_conn, id_, make_if_none=False):
 257         for row in db_conn.exec('SELECT * FROM templates WHERE id = ?', (id_,)):
 258             return cls.from_row(db_conn, row)
 259         if make_if_none:
 260             return cls(db_conn, id_)
 261         return None
 262
 263     def save(self, db_conn):
 264         cursor = db_conn.exec('REPLACE INTO templates VALUES (?) RETURNING ID', (self.id_,))
 265         if self.id_ is None:
 266             self.id_ = cursor.fetchone()[0]
 267         self.title.save(db_conn)
 268         self.default_effort.save(db_conn)
 269         self.description.save(db_conn)
 270
 271
 272
 273 ######## web stuff ############
 274
 275 class ParamsParser:
 276
 277     def __init__(self, url_query, site_cookie):
 278         self.params = parse_qs(url_query, keep_blank_values=True)
 279         self.cookie = site_cookie
 280
 281     def get(self, key, default):
 282         return self.params.get(key, [default])[0]
 283
 284     def get_cookied(self, key, default):
 285         val = self.params.get(key, [self.cookie.get(key, default)])[0]
 286         self.cookie[key] = val
 287         return val
 288
 289
 290
 291 class CookieDB:
 292
 293     def __init__(self, site, headers):
 294         from http.cookies import SimpleCookie
 295         from json import loads as json_loads
 296         self.site = site
 297         self.of_site = {}
 298         self.full = SimpleCookie(headers['Cookie']) if 'Cookie' in headers.keys() else SimpleCookie()
 299         if site in self.full.keys():
 300             self.of_site = json_loads(self.full[site].value)
 301
 302     def send_headers(self):
 303         from json import dumps as json_dumps
 304         self.full[self.site] = json_dumps(self.of_site)
 305         return [('Set-Cookie', morsel.OutputString()) for morsel in self.full.values()]
 306
 307     def reset(self):
 308         for morsel in self.full.values():
 309              morsel['expires'] = 'Thu, 01 Jan 1970 00:00:00 GMT'
 310
 311
 312
 313 class TodoHandler(BaseHTTPRequestHandler):
 314
 315     def init(self):
 316         from os.path import split as path_split
 317         from urllib.parse import urlparse
 318         parsed_url = urlparse(self.path)
 319         self.site = path_split(parsed_url.path)[1]
 320         if 0 == len(self.site):
 321             self.site = 'calendar'
 322         self.cookie = CookieDB(self.site, self.headers)
 323         self.params = ParamsParser(parsed_url.query, self.cookie.of_site)
 324         self.db_conn = TodoDBConnection(self.server.db_file)
 325
 326     def send_code_and_headers(self, code, headers, set_cookies=True):
 327         self.send_response(code)
 328         if set_cookies:
 329             [self.send_header(h[0], h[1]) for h in self.cookie.send_headers()]
 330         for fieldname, content in headers:
 331             self.send_header(fieldname, content)
 332         self.end_headers()
 333
 334     def redirect(self, url):
 335         self.send_code_and_headers(302, [('Location', url)])
 336
 337     def send_HTML(self, html, code=200):
 338         self.send_code_and_headers(code, [('Content-type', 'text/html')])
 339         self.wfile.write(bytes(html, 'utf-8'))
 340
 341     def send_fail(self, msg, code=400):
 342         html = self.server.html.get_template('msg.html').render(msg=f'Exception: {msg}')
 343         self.send_code_and_headers(code, [('Content-type', 'text/html')], set_cookies=False)
 344         self.wfile.write(bytes(html, 'utf-8'))
 345
 346     # POST routes
 347
 348     def do_POST(self):
 349         try:
 350             self.init()
 351             length = int(self.headers['content-length'])
 352             self.redir_url = '/'
 353             self.postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=1)
 354             if self.site in {'day', 'template'}:
 355                 getattr(self, f'do_POST_{self.site}')()
 356             self.db_conn.commit()
 357             self.db_conn.close()
 358             self.redirect(self.redir_url)
 359         except HandledException as msg:
 360             self.send_fail(msg)
 361
 362     def do_POST_day(self):
 363         date = self.postvars['date'][0]
 364         day = Day(date, self.postvars['comment'][0])
 365         day.save(self.db_conn)
 366
 367     def do_POST_template(self):
 368         id_ = self.params.get('id', None)
 369         tmpl = TodoTemplate.by_id(self.db_conn, id_, make_if_none=True)
 370         tmpl.title.set(self.postvars['title'][0])
 371         tmpl.default_effort.set(float(self.postvars['default_effort'][0]))
 372         tmpl.description.set(self.postvars['description'][0])
 373         tmpl.save(self.db_conn)
 374         self.redir_url = 'templates'
 375
 376     # GET routes
 377
 378     def do_GET(self):
 379         try:
 380             self.init()
 381             if self.site in {'day', 'template', 'templates', 'reset_cookie', 'calendar'}:
 382                 page = getattr(self, f'do_GET_{self.site}')()
 383             else:
 384                 page = self.do_GET_calendar()
 385             self.db_conn.close()
 386             self.send_HTML(page)
 387         except HandledException as msg:
 388             self.send_fail(msg)
 389
 390     def do_GET_reset_cookie(self):
 391         self.cookie.reset()
 392         msg = 'Cookie has been re-set!'
 393         return self.server.html.get_template('msg.html').render(msg=msg)
 394
 395     def do_GET_day(self):
 396         date = self.params.get_cookied('id', Day.todays_date())
 397         day = Day.by_date(self.db_conn, date, make_if_none=True)
 398         return self.server.html.get_template('day.html').render(day=day)
 399
 400     def do_GET_calendar(self):
 401         from_ = self.params.get_cookied('from', 'yesterday')
 402         to = self.params.get_cookied('to', '')
 403         days = Day.all(self.db_conn, ensure_betweens=True, date_range=(from_, to))
 404         return self.server.html.get_template('calendar.html').render(
 405                 days=days, from_=from_, to=to)
 406
 407     def do_GET_template(self):
 408         id_ = self.params.get('id', None)
 409         template = TodoTemplate.by_id(self.db_conn, id_, make_if_none=True)
 410         return self.server.html.get_template('template.html').render(tmpl=template)
 411
 412     def do_GET_templates(self):
 413         templates = TodoTemplate.all(self.db_conn)
 414         return self.server.html.get_template('templates.html').render(templates=templates)
 415
 416
 417
 418 def main():
 419     from http.server import HTTPServer
 420     from os import environ
 421     from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader
 422     path_todo_db = environ.get('TODO_DB')
 423     if not path_todo_db:
 424         raise HandledException('TODO_DB environment variable not set.')
 425     db_file = TodoDBFile(path_todo_db)
 426     server = HTTPServer(('localhost', HTTP_PORT), TodoHandler)
 427     server.db_file = db_file
 428     server.html = JinjaEnv(loader=JinjaFSLoader(HTML_DIR))
 429     print(f'running at http://localhost:{HTTP_PORT}')
 430     try:
 431         server.serve_forever()
 432     except KeyboardInterrupt:
 433         print('\ncaught KeyboardInterrupt, stopping server')
 434     server.server_close()
 435
 436
 437
 438 if __name__ == '__main__':
 439     try:
 440         main()
 441     except HandledException as e:
 442         e.nice_exit()