--- /dev/null
+#!/usr/bin/env python3
+
+import csv
+import datetime
+from os import (
+ makedirs,
+ path,
+ )
+import subprocess
+import sys
+import time
+
+from typing import (
+ Dict,
+ List,
+ )
+
+TimeEntries = Dict[str,List[List[str]]]
+
+TS_PATH='/home/users/.local/share/dwalde/time-sheet.csv'
+
+
+def append_entry(items: List[str]):
+ '''
+ Append a new entry to the time sheet file.
+ Takes in a list of strings typically in the form: [category,action,info...].
+ '''
+ ensure_time_dir_exists()
+ with open(TS_PATH, 'a', encoding='utf-8') as file:
+ writer = csv.writer(file)
+ writer.writerow(items)
+
+
+def ensure_time_dir_exists():
+ '''
+ Ensures that the directory containing or to contain the time sheet
+ file exists on the system.
+ Will raise any exceptions thrown by makedirs.
+ '''
+ dir_path = path.dirname(TS_PATH)
+ if not path.isdir(dir_path):
+ makedirs(dir_path, mode=0o775)
+
+
+def list_category(entries: TimeEntries, cat: str):
+ '''
+ Prints the current state of a category.
+ '''
+ last = entries[cat][-1]
+ name = entries[cat][0][1]
+ if last[0] == 'in' or last[0] == 'out':
+ print(f'{cat}:\t{name} punched {last[0]} at {last[1]}.')
+ else:
+ print(f'{cat}:\tFirst entry for {name}.')
+
+
+def list_entries(entries: TimeEntries):
+ '''
+ Print a line for each category about the current state.
+ '''
+ cats = list(entries.keys())
+ cats.sort()
+
+ for cat in cats:
+ list_category(entries, cat)
+
+
+def load_file():
+ '''
+ Loads the current state of the time sheet into memory.
+ '''
+ entries = {}
+ try:
+ with open(TS_PATH, encoding='utf-8') as file:
+ reader = csv.reader(file)
+ for row in reader:
+ if row[0] not in entries:
+ entries[row[0]] = []
+ entries[row[0]].append(row[1:])
+ except FileNotFoundError:
+ pass
+ return entries
+
+
+def save_file(entries: TimeEntries):
+ '''
+ Saves the given state back into the time sheet file.
+ TODO: Make this more fail-safe, and save to new file, then rename.
+ '''
+ with open(TS_PATH, 'w', encoding='utf-8') as file:
+ writer = csv.writer(file)
+ for category, items in entries.items():
+ for item in items:
+ writer.writerow([category] + item)
+
+
+def new_category(entries: TimeEntries, category_id: str, category_name: str):
+ '''
+ Create a new category with name and id/abbreviation.
+ '''
+ if category_id in entries:
+ print(f'{category_id} already created: {entries[category_id][0]}')
+ exit(2)
+
+ append_entry([category_id, 'category', category_name])
+
+
+def punch(entries: TimeEntries, dir: str, category: str):
+ '''
+ Attempt to add and entry to punch a category in or out.
+ _dir_ is the string 'in' or 'out'.
+ '''
+ if category not in entries:
+ print(f'Category does not exist {category}.')
+ exit(2)
+
+ last = entries[category][-1]
+
+ if last[0] == dir:
+ print(f'{category} already punched {dir}. {last}')
+ exit(3)
+ elif dir == 'out' and last[0] == 'category':
+ print(f'{category} never punched in. {last}')
+ exit(3)
+
+ now = datetime.datetime.now()
+ now = datetime.datetime.now(now.astimezone().tzinfo)
+ entry = [category, dir, now.isoformat()]
+ append_entry(entry)
+ entries[category].append(entry[1:])
+ list_category(entries, category)
+
+
+def try_punch_out(entries: TimeEntries):
+ '''
+ Checks if exactly one category is currently punched in and punches it out.
+ '''
+ category = None
+
+ for cat, items in entries.items():
+ if items[-1][0] == 'in':
+ if category is not None:
+ print(f'At least two categories punched in ({category}, {cat}), please specify.')
+ exit(3)
+ category = cat
+
+ if category is None:
+ print('No categories punched in.')
+ exit(2)
+
+ punch(entries, 'out', category)
+
+
+# Service specific
+
+def notify(title, desc):
+ subprocess.call(['notify-send', '--hint', 'int:transient:1', title, desc])
+
+
+def run_service():
+ last_notify = {}
+ while True:
+ try:
+ entries = load_file()
+ for cat, items in entries.items():
+ if items[-1][0] == 'in':
+ now = datetime.datetime.now()
+ now = datetime.datetime.now(now.astimezone().tzinfo)
+ punch = datetime.datetime.fromisoformat(items[-1][1])
+ if cat not in last_notify:
+ last_notify[cat] = punch
+ if now - last_notify[cat] > datetime.timedelta(minutes=30):
+ clock = now - punch
+ hours = int(clock.seconds/60/60)
+ minutes = int(clock.seconds/60)
+ if hours > 0:
+ timestr = f'{hours}h {minutes}m'
+ else:
+ timestr = f'{minutes}m'
+ notify('On the clock', f'{items[0][1]}: {timestr}')
+ last_notify[cat] = now
+ elif cat in last_notify:
+ del last_notify[cat]
+ time.sleep(60)
+ except KeyboardInterrupt:
+ print('Exiting.')
+ break
+
+# Main
+
+def main(args):
+ if len(args) < 2:
+ print('Missing command (daemon, new, in, out, list, print, query)')
+ exit(1)
+
+ if args[1] == 'daemon' or args[1] == 'service':
+ return run_service()
+
+ entries = load_file()
+
+ if args[1] == 'new':
+ if len(args) < 4:
+ print('Missing args for new category: <id> <name>')
+ exit(1)
+ new_category(entries, args[2], args[3])
+
+ elif args[1] == 'out':
+ if len(args) < 3:
+ try_punch_out(entries)
+ else:
+ punch(entries, 'out', args[2])
+
+ elif args[1] == 'in':
+ if len(args) < 3:
+ print(f'Missing arg for punch in: <category>')
+ exit(1)
+ punch(entries, 'in', args[2])
+
+ elif args[1] == 'pop':
+ if len(args) < 3:
+ print(f'Missing arg for pop {args[1]}: <category>')
+ exit(1)
+ if args[2] not in entries:
+ print(f'Category does not exist {args[2]}.')
+ exit(2)
+ last_entry = entries[args[2]][-1]
+ resp = input(f"Remove entry {last_entry} [y/N]?")
+ if resp.lower() == 'y':
+ entries[args[2]].pop()
+ save_file(entries)
+ list_category(entries, args[2])
+
+ elif args[1] == 'print':
+ print(entries)
+
+ elif args[1] == 'list':
+ if len(args) > 2:
+ if args[2] not in entries:
+ print(f'Category does not exist {args[2]}.')
+ exit(2)
+ list_category(entries, args[2])
+ else:
+ list_entries(entries)
+
+ elif args[1] == 'query':
+ pass
+
+
+if __name__ == "__main__":
+ main(sys.argv)
+
dev / repos / punch / commitdiff