import requests from requests.exceptions import HTTPError from requests.auth import HTTPBasicAuth from datetime import * from xdg import BaseDirectory import toml from uritools import (uricompose) from dateutil.easter import * from dateutil.tz import * from dateutil.parser import isoparse from enum import Enum from math import (copysign, ceil, floor) import calendar import argparse from copy import deepcopy import sys from sys import stderr from tabulate import tabulate from itertools import groupby, count from functools import cache, partial import backoff from pathlib import Path from collections import defaultdict import jsonpickle from hashlib import blake2s class TogglAPISection(Enum): TOGGL = '/api/v9' REPORTS = '/reports/api/v2' class TogglAPIError(Exception): def __init__(self, response, *, http_error=None): self.http_error = http_error self.response = response def __str__(self): if not self.http_error is None: return str(self.http_error) else: return self.response.text class TogglAPI(object): def __init__(self, api_token, workspace_id, client_ids): self._api_token = api_token self._workspace_id = workspace_id self._client_ids = set(map(int, client_ids.split(','))) if client_ids else None def _make_url(self, api=TogglAPISection.TOGGL, section=['me', 'time_entries', 'current'], params={}): if api is TogglAPISection.REPORTS: params.update({'user_agent': 'worktime', 'workspace_id': self._workspace_id}) api_path = api.value section_path = '/'.join(section) uri = uricompose(scheme='https', host='api.track.toggl.com', path=f"{api_path}/{section_path}", query=params) return uri def _query(self, url, method): response = self._raw_query(url, method) response.raise_for_status() return response @backoff.on_predicate( backoff.expo, factor=0.1, max_value=2, predicate=lambda r: r.status_code == 429, max_time=10, ) def _raw_query(self, url, method): headers = {'content-type': 'application/json', 'accept': 'application/json'} response = None if method == 'GET': response = requests.get(url, headers=headers, auth=HTTPBasicAuth(self._api_token, 'api_token')) elif method == 'POST': response = requests.post(url, headers=headers, auth=HTTPBasicAuth(self._api_token, 'api_token')) else: raise ValueError(f"Undefined HTTP method “{method}”") return response def entry_durations(self, start_date, *, end_date, rounding=False, client_ids): if client_ids is not None and not client_ids: return cache_dir = Path(BaseDirectory.save_cache_path('worktime')) / 'entry_durations' step = timedelta(days = 120) for req_start in (start_date + step * i for i in count()): req_start = min(req_start, end_date) req_end = min(req_start + step, end_date) if req_end <= req_start: break # if end_date > req_start + step: # req_end = datetime.combine((req_start + step).astimezone(timezone.utc).date(), time(tzinfo=timezone.utc)) # elif req_start > start_date: # req_start = datetime.combine(req_start.astimezone(timezone.utc).date(), time(tzinfo=timezone.utc)) + timedelta(days = 1) cache_path = None if req_end + timedelta(days=60) < datetime.now().astimezone(timezone.utc): cache_key = blake2s(jsonpickle.encode({ 'start': req_start, 'end': req_end, 'rounding': rounding, 'clients': client_ids, 'workspace': self._workspace_id, 'workspace_clients': self._client_ids }).encode('utf-8'), key = self._api_token.encode('utf-8')).hexdigest() cache_path = cache_dir / cache_key[:2] / cache_key[2:4] / f'{cache_key[4:]}.json' try: with cache_path.open('r', encoding='utf-8') as ch: yield from jsonpickle.decode(ch.read()) continue except FileNotFoundError: pass entries = list() params = { 'since': (req_start - timedelta(days=1)).date().isoformat(), 'until': (req_end + timedelta(days=1)).date().isoformat(), 'rounding': 'yes' if rounding else 'no', 'billable': 'yes' } if client_ids is not None: params |= { 'client_ids': ','.join(map(str, client_ids)) } for page in count(start = 1): url = self._make_url(api = TogglAPISection.REPORTS, section = ['details'], params = params | { 'page': page }) r = self._query(url = url, method='GET') if not r or not r.json(): raise TogglAPIError(r) report = r.json() for entry in report['data']: start = isoparse(entry['start']) end = isoparse(entry['end']) if start > req_end or end < req_start: continue x = min(end, req_end) - max(start, req_start) if cache_key: entries.append(x) yield x if not report['data']: break if cache_path: cache_path.parent.mkdir(parents=True, exist_ok=True) with cache_path.open('w', encoding='utf-8') as ch: ch.write(jsonpickle.encode(entries)) # res = timedelta(milliseconds=report['total_billable']) if report['total_billable'] else timedelta(milliseconds=0) # return res def get_billable_hours(self, start_date, end_date=datetime.now(timezone.utc), rounding=False): billable_acc = timedelta(milliseconds = 0) if 0 in self._client_ids: url = self._make_url(api = TogglAPISection.TOGGL, section = ['workspaces', self._workspace_id, 'clients']) r = self._query(url = url, method = 'GET') if not r or not r.json(): raise TogglAPIError(r) billable_acc += sum(self.entry_durations(start_date, end_date=end_date, rounding=rounding, client_ids=None), start=timedelta(milliseconds=0)) - sum(self.entry_durations(start_date, end_date=end_date, rounding=rounding, client_ids=frozenset(map(lambda c: c['id'], r.json()))), start=timedelta(milliseconds=0)) billable_acc += sum(self.entry_durations(start_date, end_date=end_date, rounding=rounding, client_ids=frozenset(*(self._client_ids - {0}))), start=timedelta(milliseconds=0)) return billable_acc def get_running_clock(self, now=datetime.now(timezone.utc)): url = self._make_url(api = TogglAPISection.TOGGL, section = ['me', 'time_entries', 'current']) r = self._query(url = url, method='GET') if not r or (not r.json() and r.json() is not None): raise TogglAPIError(r) if not r.json() or not r.json()['billable']: return None if self._client_ids is not None: if 'pid' in r.json() and r.json()['pid']: url = self._make_url(api = TogglAPISection.TOGGL, section = ['projects', str(r.json()['pid'])]) pr = self._query(url = url, method = 'GET') if not pr or not pr.json(): raise TogglAPIError(pr) if not pr.json(): return None if 'cid' in pr.json() and pr.json()['cid']: if pr.json()['cid'] not in self._client_ids: return None elif 0 not in self._client_ids: return None elif 0 not in self._client_ids: return None start = isoparse(r.json()['start']) return now - start if start <= now else None class Worktime(object): time_worked = timedelta() running_entry = None now = datetime.now(tzlocal()) time_pulled_forward = timedelta() now_is_workday = False include_running = True time_to_work = None force_day_to_work = True leave_days = set() excused_days = set() leave_budget = dict() time_per_day = None workdays = None @staticmethod @cache def holidays(year): holidays = dict() y_easter = datetime.combine(easter(year), time(), tzinfo=tzlocal()) # Legal holidays in munich, bavaria holidays[datetime(year, 1, 1, tzinfo=tzlocal()).date()] = 1 holidays[datetime(year, 1, 6, tzinfo=tzlocal()).date()] = 1 holidays[(y_easter+timedelta(days=-2)).date()] = 1 holidays[(y_easter+timedelta(days=+1)).date()] = 1 holidays[datetime(year, 5, 1, tzinfo=tzlocal()).date()] = 1 holidays[(y_easter+timedelta(days=+39)).date()] = 1 holidays[(y_easter+timedelta(days=+50)).date()] = 1 holidays[(y_easter+timedelta(days=+60)).date()] = 1 holidays[datetime(year, 8, 15, tzinfo=tzlocal()).date()] = 1 holidays[datetime(year, 10, 3, tzinfo=tzlocal()).date()] = 1 holidays[datetime(year, 11, 1, tzinfo=tzlocal()).date()] = 1 holidays[datetime(year, 12, 25, tzinfo=tzlocal()).date()] = 1 holidays[datetime(year, 12, 26, tzinfo=tzlocal()).date()] = 1 # TV-L holidays[datetime(year, 12, 24, tzinfo=tzlocal()).date()] = 1 holidays[datetime(year, 12, 31, tzinfo=tzlocal()).date()] = 1 # LMU holidays[(y_easter+timedelta(days=-47)).date()] = 0.5 return holidays @staticmethod def config(): config_dir = BaseDirectory.load_first_config('worktime') return toml.load(Path(config_dir) / 'worktime.toml') def ordinal_workday(self, date): start_date = datetime(date.year, 1, 1, tzinfo=tzlocal()).date() return len([1 for offset in range(0, (date - start_date).days + 1) if self.would_be_workday(start_date + timedelta(days = offset))]) def would_be_workday(self, date): return date.isoweekday() in self.workdays and date not in set(day for (day, val) in Worktime.holidays(date.year).items() if val >= 1) def is_workday(self, date, extra=True): return date in self.days_to_work or (extra and date in self.extra_days_to_work) def __init__(self, start_datetime=None, end_datetime=None, now=None, include_running=True, force_day_to_work=True, **kwargs): self.include_running = include_running self.force_day_to_work = force_day_to_work if now: self.now = now config = Worktime.config() config_dir = BaseDirectory.load_first_config('worktime') api = TogglAPI( api_token=config.get("TOGGL", {}).get("ApiToken", None), workspace_id=config.get("TOGGL", {}).get("Workspace", None), client_ids=config.get("TOGGL", {}).get("ClientIds", None) ) date_format = config.get("WORKTIME", {}).get("DateFormat", '%Y-%m-%d') self.start_date = start_datetime or datetime.strptime(config.get("WORKTIME", {}).get("StartDate"), date_format).replace(tzinfo=tzlocal()) self.end_date = end_datetime or self.now try: with open(Path(config_dir) / "reset", 'r') as reset: for line in reset: stripped_line = line.strip() reset_date = datetime.strptime(stripped_line, date_format).replace(tzinfo=tzlocal()) if reset_date > self.start_date and reset_date <= self.end_date: self.start_date = reset_date except IOError as e: if e.errno != 2: raise e def hours_per_week(day): hours_per_week = float(config.get("WORKTIME", {}).get("HoursPerWeek", 40)) for epoch in reversed(config.get("epoch", [])): from_date = None to_date = None if "from" in epoch: from_date = datetime.strptime(epoch["from"], date_format).replace(tzinfo=tzlocal()).date() if day < from_date: continue if "to" in epoch: to_date = datetime.strptime(epoch["to"], date_format).replace(tzinfo=tzlocal()).date() if day >= to_date: continue hours_per_week = float(epoch.get("HoursPerWeek", 40)) return hours_per_week self.workdays = set([int(d.strip()) for d in config.get("WORKTIME", {}).get("Workdays", '1,2,3,4,5').split(',')]) self.time_per_day = lambda day: timedelta(hours = hours_per_week(day)) / len(self.workdays) holidays = dict() leave_per_year = int(config.get("WORKTIME", {}).get("LeavePerYear", 30)) for year in range(self.start_date.year, self.end_date.year + 1): holidays |= {k: v * self.time_per_day(k) for k, v in Worktime.holidays(year).items()} leave_frac = 1 if date(year, 1, 1) < self.start_date.date(): leave_frac = (date(year + 1, 1, 1) - self.start_date.date()) / (date(year + 1, 1, 1) - date(year, 1, 1)) self.leave_budget |= {year: floor(leave_per_year * leave_frac)} try: with open(Path(config_dir) / "reset-leave", 'r') as excused: for line in excused: stripped_line = line.strip() if stripped_line: [datestr, count] = stripped_line.split(' ') day = datetime.strptime(datestr, date_format).replace(tzinfo=tzlocal()).date() if day != self.start_date.date(): continue self.leave_budget[day.year] = (self.leave_budget[day.year] if day.year in self.leave_budget else 0) + int(count) except IOError as e: if e.errno != 2: raise e for excused_kind in {'excused', 'leave'}: try: with open(Path(config_dir) / excused_kind, 'r') as excused: for line in excused: stripped_line = line.strip() if stripped_line: splitLine = stripped_line.split(' ') fromDay = toDay = None def parse_datestr(datestr): nonlocal fromDay, toDay def parse_single(singlestr): return datetime.strptime(singlestr, date_format).replace(tzinfo=tzlocal()).date() if '--' in datestr: [fromDay,toDay] = datestr.split('--') fromDay = parse_single(fromDay) toDay = parse_single(toDay) else: fromDay = toDay = parse_single(datestr) time = None if len(splitLine) == 2: [hours, datestr] = splitLine time = timedelta(hours = float(hours)) parse_datestr(datestr) else: parse_datestr(stripped_line) for day in [fromDay + timedelta(days = x) for x in range(0, (toDay - fromDay).days + 1)]: if self.end_date.date() < day or day < self.start_date.date(): continue if self.would_be_workday(day): if excused_kind == 'leave': self.leave_days.add(day) elif time is not None and time >= self.time_per_day(day): self.excused_days.add(day) holidays[day] = time if time is not None else self.time_per_day(day) except IOError as e: if e.errno != 2: raise e pull_forward = dict() start_day = self.start_date.date() end_day = self.end_date.date() try: with open(Path(config_dir) / "pull-forward", 'r') as excused: for line in excused: stripped_line = line.strip() if stripped_line: [hours, datestr] = stripped_line.split(' ') constr = datestr.split(',') for d in [start_day + timedelta(days = x) for x in range(0, (end_day - start_day).days + 1 + int(timedelta(hours = float(hours)).total_seconds() / 60 * (7 / len(self.workdays)) * 2))]: for c in constr: if c in calendar.day_abbr: if not d.strftime('%a') == c: break elif "--" in c: [fromDay,toDay] = c.split('--') if fromDay != "": fromDay = datetime.strptime(fromDay, date_format).replace(tzinfo=tzlocal()).date() if not fromDay <= d: break if toDay != "": toDay = datetime.strptime(toDay, date_format).replace(tzinfo=tzlocal()).date() if not d <= toDay: break else: if not d == datetime.strptime(c, date_format).replace(tzinfo=tzlocal()).date(): break else: if d >= self.end_date.date(): pull_forward[d] = min(timedelta(hours = float(hours)), self.time_per_day(d) - (holidays[d] if d in holidays else timedelta())) except IOError as e: if e.errno != 2: raise e self.days_to_work = dict() if pull_forward: end_day = max(end_day, max(list(pull_forward))) for day in [start_day + timedelta(days = x) for x in range(0, (end_day - start_day).days + 1)]: if day.isoweekday() in self.workdays: time_to_work = self.time_per_day(day) if day in holidays.keys(): time_to_work -= holidays[day] if time_to_work > timedelta(): self.days_to_work[day] = time_to_work self.extra_days_to_work = dict() try: with open(Path(config_dir) / "days-to-work", 'r') as extra_days_to_work_file: for line in extra_days_to_work_file: stripped_line = line.strip() if stripped_line: splitLine = stripped_line.split(' ') if len(splitLine) == 2: [hours, datestr] = splitLine day = datetime.strptime(datestr, date_format).replace(tzinfo=tzlocal()).date() self.extra_days_to_work[day] = timedelta(hours = float(hours)) else: day = datetime.strptime(stripped_line, date_format).replace(tzinfo=tzlocal()).date() self.extra_days_to_work[day] = self.time_per_day(day) except IOError as e: if e.errno != 2: raise e self.now_is_workday = self.is_workday(self.now.date()) self.time_worked = timedelta() if self.include_running: self.running_entry = api.get_running_clock(self.now) if self.running_entry: self.time_worked += self.running_entry if self.running_entry and self.include_running and self.force_day_to_work and not (self.now.date() in self.days_to_work or self.now.date() in self.extra_days_to_work): self.extra_days_to_work[self.now.date()] = timedelta() self.time_to_work = sum([self.days_to_work[day] for day in self.days_to_work.keys() if day <= self.end_date.date()], timedelta()) for day in [d for d in list(pull_forward) if d > self.end_date.date()]: days_forward = set([d for d in self.days_to_work.keys() if d >= self.end_date.date() and d < day and (not d in pull_forward or d == self.end_date.date())]) extra_days_forward = set([d for d in self.extra_days_to_work.keys() if d >= self.end_date.date() and d < day and (not d in pull_forward or d == self.end_date.date())]) days_forward = days_forward.union(extra_days_forward) extra_day_time_left = timedelta() for extra_day in extra_days_forward: day_time = max(timedelta(), self.time_per_day(extra_day) - self.extra_days_to_work[extra_day]) extra_day_time_left += day_time extra_day_time = min(extra_day_time_left, pull_forward[day]) time_forward = pull_forward[day] - extra_day_time if extra_day_time_left > timedelta(): for extra_day in extra_days_forward: day_time = max(timedelta(), self.time_per_day(extra_day) - self.extra_days_to_work[extra_day]) self.extra_days_to_work[extra_day] += extra_day_time * (day_time / extra_day_time_left) hours_per_day_forward = time_forward / len(days_forward) if len(days_forward) > 0 else timedelta() days_forward.discard(self.end_date.date()) self.time_pulled_forward += time_forward - hours_per_day_forward * len(days_forward) if self.end_date.date() in self.extra_days_to_work: self.time_pulled_forward += self.extra_days_to_work[self.end_date.date()] self.time_to_work += self.time_pulled_forward self.time_worked += api.get_billable_hours(self.start_date, self.now, rounding = config.get("WORKTIME", {}).get("rounding", True)) def format_days(worktime, days, date_format=None): if not date_format: config = Worktime.config() date_format = config.get("WORKTIME", {}).get("DateFormat", '%Y-%m-%d') groups = list(map(lambda kv: list(kv[1]), groupby( map( lambda kv: list(map(lambda t: t[1], kv[1])), groupby(enumerate(days), lambda kv: kv[0] - worktime.ordinal_workday(kv[1])) ), lambda days: frozenset(map(lambda day: (day.isocalendar().year, day.isocalendar().week), days)) ))) def format_group(group): if len(group) > 1: return group[0].strftime(date_format) + '…' + group[-1].strftime(date_format) else: return group[0].strftime(date_format) return ', '.join(map(lambda group: ','.join(map(format_group, group)), groups)) def worktime(**args): worktime = Worktime(**args) def format_worktime(worktime): def difference_string(difference): total_minutes_difference = round(difference / timedelta(minutes = 1)) (hours_difference, minutes_difference) = divmod(abs(total_minutes_difference), 60) sign = '' if total_minutes_difference >= 0 else '-' difference_string = f"{sign}" if hours_difference != 0: difference_string += f"{hours_difference}h" if hours_difference == 0 or minutes_difference != 0: difference_string += f"{minutes_difference}m" return difference_string difference = worktime.time_to_work - worktime.time_worked total_minutes_difference = 5 * ceil(difference / timedelta(minutes = 5)) if worktime.running_entry and abs(difference) < timedelta(days = 1) and (total_minutes_difference > 0 or abs(worktime.running_entry) >= abs(difference)) : clockout_time = worktime.now + difference clockout_time += (5 - clockout_time.minute % 5) * timedelta(minutes = 1) clockout_time = clockout_time.replace(second = 0, microsecond = 0) if total_minutes_difference >= 0: difference_string = difference_string(total_minutes_difference * timedelta(minutes = 1)) return f"{difference_string}/{clockout_time:%H:%M}" else: difference_string = difference_string(abs(total_minutes_difference) * timedelta(minutes = 1)) return f"{clockout_time:%H:%M}/{difference_string}" else: if worktime.running_entry: difference_string = difference_string(abs(total_minutes_difference) * timedelta(minutes = 1)) indicator = '↓' if total_minutes_difference >= 0 else '↑' # '\u25b6' return f"{indicator}{difference_string}" else: difference_string = difference_string(total_minutes_difference * timedelta(minutes = 1)) if worktime.now_is_workday: return difference_string else: return f"({difference_string})" if worktime.time_pulled_forward >= timedelta(minutes = 15): worktime_no_pulled_forward = deepcopy(worktime) worktime_no_pulled_forward.time_to_work -= worktime_no_pulled_forward.time_pulled_forward worktime_no_pulled_forward.time_pulled_forward = timedelta() difference_string = format_worktime(worktime) difference_string_no_pulled_forward = format_worktime(worktime_no_pulled_forward) print(f"{difference_string_no_pulled_forward}…{difference_string}") else: print(format_worktime(worktime)) def time_worked(now, **args): then = now.replace(hour = 0, minute = 0, second = 0, microsecond = 0) if now.time() == time(): now = now + timedelta(days = 1) then = Worktime(**dict(args, now = then)) now = Worktime(**dict(args, now = now)) worked = now.time_worked - then.time_worked if args['do_round']: total_minutes_difference = 5 * ceil(worked / timedelta(minutes = 5)) (hours_difference, minutes_difference) = divmod(abs(total_minutes_difference), 60) sign = '' if total_minutes_difference >= 0 else '-' difference_string = f"{sign}" if hours_difference != 0: difference_string += f"{hours_difference}h" if hours_difference == 0 or minutes_difference != 0: difference_string += f"{minutes_difference}m" clockout_time = None clockout_difference = None if then.now_is_workday or now.now_is_workday: target_time = max(then.time_per_day(then.now.date()), now.time_per_day(now.now.date())) if then.time_per_day(then.now.date()) and now.time_per_day(now.now.date()) else (then.time_per_day(then.now.date()) if then.time_per_day(then.now.date()) else now.time_per_day(now.now.date())); difference = target_time - worked clockout_difference = 5 * ceil(difference / timedelta(minutes = 5)) clockout_time = now.now + difference clockout_time += (5 - clockout_time.minute % 5) * timedelta(minutes = 1) clockout_time = clockout_time.replace(second = 0, microsecond = 0) if now.running_entry and clockout_time and clockout_difference >= 0: print(f"{difference_string}/{clockout_time:%H:%M}") else: print(difference_string) else: print(worked) def diff(now, **args): now = now.replace(hour = 0, minute = 0, second = 0, microsecond = 0) then = now - timedelta.resolution then = Worktime(**dict(args, now = then, include_running = False)) now = Worktime(**dict(args, now = now, include_running = False)) print(now.time_to_work - then.time_to_work) def holidays(year, table_format, **args): config = Worktime.config() date_format = config.get("WORKTIME", {}).get("DateFormat", '%Y-%m-%d') table_data = [] holidays = Worktime.holidays(year) for k, v in holidays.items(): kstr = k.strftime(date_format) table_data += [[kstr, v]] print(tabulate(table_data, tablefmt=table_format, headers=["Date", "Proportion"] if table_format != 'plain' else None)) def leave(year, table, table_format, **args): def_year = datetime.now(tzlocal()).year worktime = Worktime(**dict(**args, end_datetime = datetime(year = (year if year else def_year) + 1, month = 1, day = 1, tzinfo=tzlocal()) - timedelta(microseconds=1))) config = Worktime.config() date_format = config.get("WORKTIME", {}).get("DateFormat", '%Y-%m-%d') leave_expires = config.get("WORKTIME", {}).get("LeaveExpires", None) if leave_expires: leave_expires = datetime.strptime(leave_expires, '%m-%d').date() days = [worktime.start_date.date() + timedelta(days = x) for x in range(0, (worktime.end_date.date() - worktime.start_date.date()).days + 1)] leave_budget = deepcopy(worktime.leave_budget) year_leave_budget = deepcopy(worktime.leave_budget) if year else None years = sorted(leave_budget.keys()) for day in sorted(worktime.leave_days): if not worktime.would_be_workday(day): continue for iyear in years: if day > leave_expires.replace(year = iyear + 1): continue if leave_budget[iyear] <= 0: continue leave_budget[iyear] -= 1 if year_leave_budget and day.year < year: year_leave_budget[iyear] -= 1 break else: print(f'Unaccounted leave: {day}', file=stderr) if table and year: table_data = [] leave_days = sorted([day for day in worktime.leave_days if day.year == year and worktime.would_be_workday(day)]) count = 0 for _, group in groupby(enumerate(leave_days), lambda kv: kv[0] - worktime.ordinal_workday(kv[1])): group = list(map(lambda kv: kv[1], group)) for day in group: for iyear in years: if day > leave_expires.replace(year = iyear + 1): continue if year_leave_budget[iyear] <= 0: continue year_leave_budget[iyear] -= 1 break next_count = count + len(group) if len(group) > 1: table_data.append([count, group[0].strftime('%m–%d') + '…' + group[-1].strftime('%m–%d'), len(group), sum(year_leave_budget.values())]) else: table_data.append([count, group[0].strftime('%m–%d'), len(group), sum(year_leave_budget.values())]) count = next_count print(tabulate(table_data, tablefmt=table_format, headers=["Running Count", "Leave Days", "# of Leave Days", "# of Leave Days Left"] if table_format != 'plain' else None)) elif table: table_data = [] for year, offset in leave_budget.items(): leave_days = sorted([day for day in worktime.leave_days if day.year == year]) would_be_workdays = [day for day in days if day.year == year and worktime.would_be_workday(day)] table_data += [[year, offset, f"{len(leave_days)}/{len(list(would_be_workdays))}", format_days(worktime, leave_days, date_format='%m–%d')]] print(tabulate(table_data, tablefmt=table_format, headers=["Year", "Δ of Leave Days", " # of Leave Days\n/ Pot. Workdays", "Leave Days"] if table_format != 'plain' else None)) else: print(leave_budget[year if year else def_year]) def classification(classification_name, table, table_format, **args): worktime = Worktime(**args) config = Worktime.config() date_format = config.get("WORKTIME", {}).get("DateFormat", '%Y-%m-%d') config_dir = BaseDirectory.load_first_config('worktime') days = [worktime.start_date.date() + timedelta(days = x) for x in range(0, (worktime.end_date.date() - worktime.start_date.date()).days + 1)] classify_excused = config.get("day-classification", {}).get(classification_name, {}).get("classify-excused", False) classify_non_workdays = config.get("day-classification", {}).get(classification_name, {}).get("classify-non-workdays", classify_excused) if classify_excused and not classify_non_workdays: print('classify_excused but not classify_non_workdays', file=stderr) year_classification = defaultdict(dict) year_offset = defaultdict(lambda: 0) extra_classified = dict() classification_files = { Path(config_dir) / classification_name: True, Path(config_dir) / f"extra-{classification_name}": True, Path(config_dir) / f"not-{classification_name}": False, } for path, val in classification_files.items(): try: with path.open('r') as fh: for line in fh: stripped_line = line.strip() if stripped_line: fromDay = toDay = None def parse_single(singlestr): return datetime.strptime(singlestr, date_format).replace(tzinfo=tzlocal()).date() if '--' in stripped_line: [fromDay,toDay] = stripped_line.split('--') fromDay = parse_single(fromDay) toDay = parse_single(toDay) else: fromDay = toDay = parse_single(stripped_line) for day in [fromDay + timedelta(days = x) for x in range(0, (toDay - fromDay).days + 1)]: if day in extra_classified: print(f'Conflicting classification: {day}', file=stderr) extra_classified[day] = val except IOError as e: if e.errno != 2: raise e extra_classified_used = set() for day in days: if not classify_excused and worktime.is_workday(day, extra=False) == classify_non_workdays: continue if classify_excused and day not in worktime.excused_days: continue classification_days = set() for default_classification in reversed(config.get("day-classification", {}).get(classification_name, {}).get("default", [])): from_date = None to_date = None if "from" in default_classification: from_date = datetime.strptime(default_classification["from"], date_format).replace(tzinfo=tzlocal()).date() if day < from_date: continue if "to" in default_classification: to_date = datetime.strptime(default_classification["to"], date_format).replace(tzinfo=tzlocal()).date() if day >= to_date: continue classification_days = set([int(d.strip()) for d in default_classification.get("days", "").split(',') if d.strip()]) default_classification = day.isoweekday() in classification_days override = None if day in extra_classified: override = extra_classified[day] if override != default_classification: extra_classified_used.add(day) year_offset[day.year] += 1 if override else -1 year_classification[day.year][day] = override if override is not None else default_classification if set(extra_classified.keys()) - set(extra_classified_used): print(f'Unused manual classification(s): {set(extra_classified.keys()) - set(extra_classified_used)}', file=stderr) if not table: print(sum(year_offset.values())) else: table_data = [] for year in sorted(year_classification.keys()): row_data = [year, year_offset[year]] classified = [day for day, classified in year_classification[year].items() if classified] count_would_be_workdays = len([1 for day in days if day.year == year and (classify_excused or (worktime.would_be_workday(day) and day not in worktime.leave_days) != classify_non_workdays) and (not classify_excused or day in worktime.excused_days)]) if len(year_classification[year]) != count_would_be_workdays: row_data.append(f"{len(classified)}/{len(year_classification[year])}/{count_would_be_workdays}") else: row_data.append(f"{len(classified)}/{len(year_classification[year])}") row_data.append(format_days(worktime, classified, date_format='%m–%d')) table_data.append(row_data) print(tabulate( table_data, tablefmt=table_format, headers=["Year", "Running Difference", f" # of {classification_name.title()} Days\n / # of Classified Days\n(/ # of Pot. Classifiable Days)", f"{classification_name.title()} Days"] if table_format != 'plain' else None, maxcolwidths=[None, None, None, 50] if table_format != 'plain' else None )) def main(): def isotime(s): return datetime.fromisoformat(s).replace(tzinfo=tzlocal()) config = Worktime.config() parser = argparse.ArgumentParser(prog = "worktime", description = 'Track worktime using toggl API') parser.add_argument('--time', dest = 'now', metavar = 'TIME', type = isotime, help = 'Time to calculate status for (default: current time)', default = datetime.now(tzlocal())) parser.add_argument('--start', dest = 'start_datetime', metavar = 'TIME', type = isotime, help = 'Time to calculate status from (default: None)', default = None) parser.add_argument('--no-running', dest = 'include_running', action = 'store_false') parser.add_argument('--no-force-day-to-work', dest = 'force_day_to_work', action = 'store_false') subparsers = parser.add_subparsers(help = 'Subcommands') parser.set_defaults(cmd = worktime) time_worked_parser = subparsers.add_parser('time_worked', aliases = ['time', 'worked', 'today']) time_worked_parser.add_argument('--no-round', dest = 'do_round', action = 'store_false') time_worked_parser.set_defaults(cmd = time_worked) diff_parser = subparsers.add_parser('diff') diff_parser.set_defaults(cmd = diff) holidays_parser = subparsers.add_parser('holidays') holidays_parser.add_argument('--table-format', dest='table_format', type=str, default='fancy_grid') holidays_parser.add_argument('year', metavar = 'YEAR', type = int, help = 'Year to evaluate holidays for (default: current year)', default = datetime.now(tzlocal()).year, nargs='?') holidays_parser.set_defaults(cmd = holidays) leave_parser = subparsers.add_parser('leave') leave_parser.add_argument('year', metavar = 'YEAR', type = int, help = 'Year to evaluate leave days for (default: current year)', default = None, nargs='?') leave_parser.add_argument('--table', action = 'store_true') leave_parser.add_argument('--table-format', dest='table_format', type=str, default='fancy_grid') leave_parser.set_defaults(cmd = leave) for classification_name in config.get('day-classification', {}).keys(): classification_parser = subparsers.add_parser(classification_name) classification_parser.add_argument('--table', action = 'store_true') classification_parser.add_argument('--table-format', dest='table_format', type=str, default='fancy_grid') classification_parser.set_defaults(cmd = partial(classification, classification_name=classification_name)) args = parser.parse_args() args.cmd(**vars(args)) if __name__ == "__main__": sys.exit(main())