hamster-tally

#!/usr/bin/env python

import itertools as it, operator as op, functools as ft
import pathlib as pl, datetime as dt, subprocess as sp, collections as cs
import os, sys, re, logging, math, csv


class LogMessage:
	def __init__(self, fmt, a, k): self.fmt, self.a, self.k = fmt, a, k
	def __str__(self): return self.fmt.format(*self.a, **self.k) if self.a or self.k else self.fmt

class LogStyleAdapter(logging.LoggerAdapter):
	def __init__(self, logger, extra=None): super().__init__(logger, extra or dict())
	def log(self, level, msg, *args, **kws):
		if not self.isEnabledFor(level): return
		log_kws = {} if 'exc_info' not in kws else dict(exc_info=kws.pop('exc_info'))
		msg, kws = self.process(msg, kws)
		self.logger._log(level, LogMessage(msg, args, kws), (), **log_kws)

err_fmt = lambda err: f'[{err.__class__.__name__}] {err}'
get_logger = lambda name: LogStyleAdapter(logging.getLogger(f'ht.{name}'))

class adict(dict):
	def __init__(self, *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.__dict__ = self


_td_days = dict( y=365.25, yr=365.25,
	year=365.25, mo=30.5, month=30.5, w=7, week=7, d=1, day=1 )
_td_s = dict( h=3600, hr=3600, hour=3600,
	m=60, min=60, minute=60, s=1, sec=1, second=1 )
def _td_re():
	sub_sort = lambda d: sorted(
		d.items(), key=lambda kv: (kv[1], len(kv[0])), reverse=True )
	ts_re = ['^'] + [
		r'(?P<{0}>\d*{0}s?\s*)?'.format(k)
		for k, v in it.chain.from_iterable(
			map(sub_sort, [_td_days, _td_s]) ) ] + ['$']
	return re.compile(''.join(ts_re), re.I | re.U)
_td_re = _td_re()

def _td_parse(td_str):
	if ':' in td_str:
		t = dt.datetime.strptime(td_str, '%H:%M:%S')
		return dt.timedelta(hours=t.hour, minutes=t.minute, seconds=t.second)
	try: return dt.timedelta(seconds=float(td_str))
	except ValueError: pass
	if not (m := _td_re.search(td_str)) or not any(m.groups()):
		raise ValueError(td_str) from None
	delta, parse_int = list(), lambda v: int(''.join(c for c in v if c.isdigit()) or 1)
	for units in _td_days, _td_s:
		val = 0
		for k, v in units.items():
			try:
				if not m.group(k): continue
				n = parse_int(m.group(k))
			except IndexError: continue
			val += n * v
		delta.append(val)
	return dt.timedelta(*delta)

def td_parse(td_str, to_float=True):
	delta = _td_parse(td_str)
	return delta if not to_float else delta.total_seconds()

def td_repr( ts, ts0=None, units_max=2, units_res=None,
		_units=dict( h=3600, m=60, s=1,
			y=365.25*86400, mo=30.5*86400, w=7*86400, d=1*86400 ) ):
	delta = ts if ts0 is None else (ts - ts0)
	if isinstance(delta, dt.timedelta): delta = delta.total_seconds()
	res, s, n_last = list(), abs(delta), units_max - 1
	units = sorted(_units.items(), key=lambda v: v[1], reverse=True)
	for unit, unit_s in units:
		val = math.floor(s / unit_s)
		if not val:
			if units_res == unit: break
			continue
		if len(res) == n_last or units_res == unit:
			val, n_last = round(s / unit_s), True
		res.append(f'{val:.0f}{unit}')
		if n_last is True: break
		s -= val * unit_s
	return ' '.join(res) if res else '<1s'


class LogError(Exception): pass

def dd_week(dd1):
	dd1 -= dt.timedelta(days=dd1.weekday())
	dd2 = dd1 + dt.timedelta(days=6)
	return dd1, dd2

log_fn_months = ( 'January February March April May'
	' June July August September October November December' ).split()
def log_fn_fmt_vars(dd1, dd2):
	dd, dd1_str = dd1, f'{dd1.day:02d}'
	if dd1.month != dd2.month: # week split between two months
		if dd2.day < 4: dd = dd1
		else: dd, dd1_str = dd2, f'-{dd1_str}'
	vs = adict( year=f'{dd.year:04d}',
		mo=f'{dd.month:02d}', dd1=dd1_str, dd2=f'{dd2.day:02d}' )
	vs.mo_name = f'{vs.mo}_{log_fn_months[dd.month-1]}'
	return vs

def log_fn_fmt_re(log_fmt):
	log_fmt_vars = dict.fromkeys(
		log_fn_fmt_vars(dt.date.today(), dt.date.today() ).keys() )
	log_fmt = re.escape(log_fmt.replace('{{', '\ue010').replace('}}', '\ue020'))\
		.replace(r'\{', '{').replace(r'\}', '}').replace('\ue010', '{{').replace('\ue020', '}}')
	log_fmt_vars.update( year=r'\d{4}', mo=r'\d{2}',
		mo_name=r'\d\d_\w+', dd1=r'-?\d\d', dd2=r'\d\d' )
	if not all(log_fmt_vars.values()): raise AssertionError(log_fmt_vars)
	return re.compile(f'^{log_fmt.format(**log_fmt_vars)}$')

def log_total_td(lines): return sum(map(op.attrgetter('td'), lines))
def log_total_line(lines):
	dd1, dd2 = dd_week(lines[-1].date)
	return f'-- total [ {dd1.isoformat()} - {dd2.isoformat()} ] :: {td_repr(log_total_td(lines))}'

class LogFile(list): td_total = None
log_line = cs.namedtuple('LogLine', 'fn n date ts1 ts2 td desc')
def log_reparse(fn):
	lines, log = LogFile(), get_logger('reparse')
	for n, line in enumerate(pl.Path(fn).read_text().splitlines(), 1):
		if not line.strip() or line.startswith('-- day/date -- '): continue
		if line.startswith('-- total '):
			lines.td_total = log_total_td(lines)
			if line != log_total_line(lines):
				raise LogError(f'Unrecognized totals-line format [ {fn}:{n} ]: {line!r}')
			continue
		elif lines.td_total is not None:
			raise LogError(f'Timespan data after totals-line [ {fn}:{n} ]: {line!r}')
		try:
			dts, tss, td, desc = map(str.strip, line.split('::', 3))
			dts, td = dt.date.fromisoformat(dts.split()[-1]), td_parse(td)
			ts1, ts2 = (dt.time.fromisoformat(ts.strip()) for ts in tss.split('-', 1))
			lines.append(log_line(fn, n, dts, ts1, ts2, td, desc))
		except Exception as err:
			log.error(f'Unrecognized log-line format [ {fn}:{n} ]: {line!r} - {err_fmt(err)}')
			continue
	if not lines: raise LogError(f'Empty existing log-file [ {fn} ]')
	return lines

def git_env_parse(**env_defaults):
	git_env, git_pre, log = dict(), dict(), get_logger('git-env')
	for k in 'name', 'email': # lookup values in author/comitter env or gitconfig
		for sk in 'author', 'committer':
			res = os.environ.get(f'GIT_{sk}_{k}'.upper(), '').strip()
			if not res:
				res = sp.run(['git', 'config', '--get', f'{sk}.{k}'], capture_output=True)
				res = res.stdout.decode().strip() if not res.returncode and res.stdout else None
			if res: git_env[k] = res; break
	for k in 'name', 'email': # check user.name/user.email
		if res := git_env.get(k): git_pre[k] = res; continue
		res = sp.run(['git', 'config', '--get', f'user.{k}'], capture_output=True)
		if not res.returncode and res.stdout:
			git_pre[k] = git_env[k] = res.stdout.decode().strip()
		elif res := env_defaults.get(k): git_env[k] = res
	for k in 'name', 'email':
		res = git_env.pop(k, None)
		for sk in 'author', 'committer':
			ek = f'GIT_{sk}_{k}'.upper()
			if res := os.environ.get(ek, '').strip() or res: git_env[ek] = res
			else: log.error(f'Missing git config/env {sk}.{k} value, with no default in cli options')
	if git_pre: log.debug('git presets: {}', ' '.join(f'{k}={v!r}' for k,v in git_pre.items()))
	return git_env


def main(args=None):
	import argparse, textwrap, re
	dd = lambda text: re.sub( r' \t+', ' ',
		textwrap.dedent(text).strip('\n') + '\n' ).replace('\t', '  ')
	parser = argparse.ArgumentParser( usage='%(prog)s [options]',
		formatter_class=argparse.RawTextHelpFormatter, description=dd('''
			Tool to t̶a̶l̶l̶y̶ ̶t̶h̶e̶ ̶h̶a̶m̶s̶t̶e̶r̶s̶ copy and commit daily time-tracking
				entries from Hamster app to a weekly logs in a git repository.
			Uses CLI "hamster export" tool, needs to have dbus access to hamster-service.
			Intended to be run at the start of a new day from crontab or such.
			If no earlier logs are found in the repo, tallying will start from current week.
			Project Hamster time tracking toolkit: https://github.com/projecthamster'''))

	group = parser.add_argument_group('Log filename/path options')
	group.add_argument('-f', '--log-file-fmt', metavar='git-path-fmt',
		default='activity/{year}/{mo}/week.{dd1}-{dd2}.log', help=dd('''
			Format for path to generated weekly activity-log files.
			Should be python str.format template, optionally using
				following keys: {}.
			Default: %(default)s'''.format(', '.join(
				log_fn_fmt_vars(dt.date.today(), dt.date.today()).keys() ))))
	group.add_argument('--symlink-last-log',
		metavar='git-path', default='activity.latest-week.log', help=dd('''
			Git repo path of a symlink to maintain, pointing to a latest log file.
			Special value "-" disables creating/updating it. Defauilt: %(default)s'''))
	group.add_argument('--symlink-last-logdir',
		metavar='git-path', default='activity.latest-month', help=dd('''
			Git repo path of a symlink to maintain, pointing to a directory with a latest log file.
			Ignored if --log-file-fmt has no slashes, i.e. when logs are created in repo root.
			Special value "-" disables creating/updating it. Defauilt: %(default)s'''))

	group = parser.add_argument_group('Git-related options')
	group.add_argument('-r', '--git-repo', metavar='path',
		help='Path to git repository to commit updates into. Default: current dir.')
	group.add_argument('--git-name',
		metavar='name', default='hamster-tally', help=dd('''
			Name to use for GIT_AUTHOR_NAME and
				GIT_COMMITTER_NAME env vars, if these are not set in
				neither environment nor in user/author/comitter.name git-config values.
			Special value "-" can be used to disable setting these. Default: %(default)s.'''))
	group.add_argument('--git-email',
		metavar='email', default=f'hamster-tally@{os.uname().nodename}',
		help='Same as --git-name above, but for GIT_AUTHOR_EMAIL/GIT_COMMITTER_EMAIL.')
	group.add_argument('--git-commit-type', default='always',
		metavar='(always|totals|never)', choices='always totals never'.split(), help=dd('''
			When/whether to commit added files/changes into git repo.
			One of: always - commit any changes, totals - commit when any logs are done, never.'''))
	group.add_argument('-o', '--git-commit-opts', metavar='opts',
		help='Space-separated extra options for "git commit" command.')
	group.add_argument('-p', '--git-push', action='store_true', help='Run "git push" after commit.')
	group.add_argument('-P', '--git-push-remote', metavar='remote', help=dd('''
		Run "git push <remote>" after commit(s).
		Same as -p/--git-push, but to use specific git-remote name.'''))

	group = parser.add_argument_group('Misc/debug opts')
	group.add_argument('--debug', action='store_true', help='Verbose operation mode.')

	opts = parser.parse_args(sys.argv[1:] if args is None else args)

	logging.basicConfig(
		level=logging.DEBUG if opts.debug else logging.WARNING,
		format='{name} {levelname:5} :: {message}', style='{' )
	log = get_logger('main')

	dd1, dd2 = dd_week(dt.date.today()) # current week date-span
	dt1 = dt.datetime(dd1.year, dd1.month, dd1.day)
	dt2 = dt.datetime(dd2.year, dd2.month, dd2.day) + dt.timedelta(1) # dd2 is Sun

	if opts.git_repo: os.chdir(opts.git_repo)
	git_path = pl.Path('.').resolve(True)
	git_cmd = ft.partial(sp.run, check=True, env=dict( GIT_EDITOR='/bin/false',
		**git_env_parse(name=opts.git_name.strip('-'), email=opts.git_email.strip('-')) ))
	git_push = opts.git_push or opts.git_push_remote
	def git_path_rel(p, res=True):
		p_res = git_path / p
		if res: p_res = p_res.resolve()
		if not p_res.is_relative_to(git_path):
			raise AssertionError(f'Specified path outside of git repo: {p}')
		return pl.Path(str(p_res)[len(str(git_path))+1:])
	link_last_log = opts.symlink_last_log and git_path_rel(opts.symlink_last_log, False)
	link_last_log_dir = opts.symlink_last_logdir and git_path_rel(opts.symlink_last_logdir, False)

	# Find last logged timestamp to use as start-date in hamster-export, if any
	log_re = log_fn_fmt_re(opts.log_file_fmt)
	res = git_cmd(['git', 'ls-files', '-z'], stdout=sp.PIPE)
	if log_list := sorted(map( git_path_rel,
			filter(log_re.search, res.stdout.decode().split('\0')) )):
		log.debug('Detected previous log-file: {}', log_last := log_list[-1])
		line = log_reparse(log_last)[-1]
		dt1 = dt.datetime.combine(line.date, line.ts2) + dt.timedelta(minutes=1)
		if line.ts2 <= line.ts1: dt1 += dt.timedelta(days=1)
		log.debug('Using start-date after last log-file line: {}', dt1)

	# Request and parse TSV of events in dt1-dt2 timespan
	ts1, ts2 = ( dt1.strftime('%Y-%m-%d %H:%M'),
		(dt2 + dt.timedelta(days=1)).strftime('%Y-%m-%d %H:%M') )
	log.debug('Hamster tsv-query: {} - {}', ts1, ts2)
	res = sp.run(['hamster', 'export', 'tsv', ts1, ts2], check=True, stdout=sp.PIPE)
	tsv = list(csv.reader(res.stdout.decode().splitlines(), 'excel-tab'))
	# Expected fields: activity, start, end, duration, category, description, tags
	tsv_line = cs.namedtuple('EvTSV', list(k.split(' ', 1)[0] for k in tsv[0]))
	tsv = list( tsv_line(*line) for line in
		(list(s.strip() for s in line) for line in tsv[1:]) if any(line) )
	log.debug('Processed hamster tsv events: {}', len(tsv))

	# Convert TSV entries to new log-events
	new_evs, new_ev = list(), cs.namedtuple('Ev', 'date ts1 ts2 td desc')
	for tev in tsv:
		if not tev.end: continue # activity-span is not running right now, log later when done
		ts1, ts2 = map(dt.datetime.fromisoformat, [tev.start, tev.end])
		if ts2 <= dt1: continue # in case export returns old events
		desc = tev.activity.replace(';', ',') # hamster >3.02 treats commas specially #753
		if tev.description.strip(): desc += f' - {tev.description}'
		evs, desc = None, desc.replace('\n', ' // ')
		if ts1.date() != ts2.date(): # starts/ends on different days, split if ts2>dt2
			if ts2.date() - ts1.date() != dt.timedelta(days=1): raise AssertionError(ev)
			if ts2 > dt2: # can still log as one ev otherwise
				evs = [
					new_ev( ts1.date(), ts1.time(), dt.time(23, 59),
						dt.timedelta(hours=23 - ts1.hour, minutes=60 - ts1.minute), desc ),
					new_ev( ts2.date(), dt.time(0, 0),
						ts2.time(), ts2 - ts2.replace(hour=0, minute=0), desc ) ]
		if not evs:
			evs = [new_ev( ts1.date(), ts1.time(), ts2.time(),
				dt.timedelta(minutes=round(float(tev.duration))), desc )]
		new_evs.extend(evs)

	# Append new log-events to any relevant log files
	weekday_names = 'mon tue wed thu fri sat sun'.title().split()
	log_paths_upd = set()
	for ev in new_evs:
		log_fmt_vars = log_fn_fmt_vars(*dd_week(ev.date))
		log_path = git_path_rel(opts.log_file_fmt.format(**log_fmt_vars))
		log.debug( 'New log-event [ {} ]: {}', log_path,
			' '.join(f'{k}={f"[{vs}]" if " " in (vs:=str(v)) else vs}' for k,v in ev._asdict().items()) )
		log_path.parent.mkdir(parents=True, exist_ok=True)
		with log_path.open('a') as dst:
			if not dst.tell():
				dst.write('-- day/date -- :: - time span - ::  delta  :: activity/description\n\n')
			dst.write(
				f'{weekday_names[ev.date.weekday()]} {ev.date.isoformat()}'
				f' :: {ev.ts1.strftime("%H:%M")} - {ev.ts2.strftime("%H:%M")}'
				f' :: {td_repr(ev.td):>7} :: {ev.desc}\n' )
		log_paths_upd.add(log_path)

	# Check last/updated closed-week logs for adding final "totals" line there
	dd_now, totals = dt.date.today(), 0
	for log_path in log_paths_upd.union(log_list[-2:]):
		lines = log_reparse(log_path)
		if lines.td_total: continue
		ldd1, ldd2 = dd_week(lines[-1].date)
		if dd_now <= ldd2: continue
		log_paths_upd.add(log_path); totals += 1; line = log_total_line(lines)
		log.debug('Adding log totals-line [ {} ]: {}', log_path, line)
		with log_path.open('a') as dst: dst.write(f'\n{line}\n')
	log_list = sorted(log_paths_upd.union(log_list))

	# git-add updated logfiles, update symlinks, if enabled
	if log_paths_upd: git_cmd(['git', 'add', '--', *log_paths_upd])
	if log_last := log_list and log_list[-1]:
		if link_last_log and link_last_log.resolve() != log_last.resolve():
			if link_last_log.exists() and not link_last_log.is_symlink():
				raise AssertionError(f'Exising --symlink-* path is not a symlink: {link_last_log}')
			link_last_log.unlink(missing_ok=True)
			os.symlink( link_dst :=
				('../'*str(link_last_log).count('/')) + str(log_last), link_last_log )
			log.debug('Updated last-log symlink: {} -> {}', link_last_log, link_dst)
			git_cmd(['git', 'add', '-f', '--', link_last_log])
		if ( link_last_log_dir and log_last.parent.resolve() != git_path
				and link_last_log_dir.resolve() != log_last.parent.resolve() ):
			if link_last_log_dir.exists() and not link_last_log_dir.is_symlink():
				raise AssertionError(f'Exising --symlink-* path is not a symlink: {link_last_log_dir}')
			link_last_log_dir.unlink(missing_ok=True)
			os.symlink( link_dst :=
				('../'*str(link_last_log_dir).count('/')) + str(log_last.parent), link_last_log_dir )
			log.debug('Updated last-log-dir symlink: {} -> {}', link_last_log, link_dst)
			git_cmd(['git', 'add', '-f', '--', link_last_log_dir])

	# git-commit/push, if enabled
	if log_paths_upd:
		git_ct_totals = opts.git_commit_type == 'totals' and totals
		if opts.git_commit_type == 'always' or git_ct_totals:
			upd_msg = f'Scripted update-commit [ week {dd1.isoformat()} - {dd2.isoformat()} ]'
			if git_ct_totals: upd_msg += f': {totals} finished logs'
			else:
				upd_metrics = list()
				if new_evs: upd_metrics.append(f'{len(new_evs)} new timespan(s)')
				if totals: upd_metrics.append(f'{totals} finished log(s)')
				if log_paths_upd: upd_metrics.append(f'{len(log_paths_upd)} updated file(s)')
				if upd_metrics: upd_msg += '\n\n' + f'Changes: {", ".join(upd_metrics)}'
			log.debug( 'Creating git commit (when={} push={}) [ {} - {} ]',
				opts.git_commit_type, 'ny'[bool(git_push)], dd1.isoformat(), dd2.isoformat() )
			git_cmd(['git', 'commit', '--no-edit', *opts.git_commit_opts.split(), '-aqm', upd_msg])
			if git_push:
				git_push_remote = [] if git_push is True else [git_push]
				git_cmd(['git', 'push', '-q'] + git_push_remote)

if __name__ == '__main__': sys.exit(main())