on-modify-timeclock.py

#!/usr/bin/env python3
import os as _os_timing, time as _time_module
if _os_timing.environ.get('TW_TIMING'):
    import atexit as _atexit
    _t0 = _time_module.perf_counter()

    def _report_timing(_f=__file__):
        elapsed = (_time_module.perf_counter() - _t0) * 1000
        import os.path as _osp
        print(f"[timing] {_osp.basename(_f)}: {elapsed:.1f}ms", file=__import__('sys').stderr)

    _atexit.register(_report_timing)


# ============================================================================
# DEBUG VERSION - Auto-generated by make-awesome.py
# ============================================================================

import os
import sys
from pathlib import Path
from datetime import datetime
#
# Writes task start/stop times to a timelog formatted file.
# You might need to adjust LEDGERFILE, or set the TIMELOG environment variable.
#
# Example reports, after using start/stop on a task:
# ledger -f /path/to/timelog.ledger print
# ledger -f /path/to/timelog.ledger register
#
# Projects, tags, and UUIDs are fully supported and queryable from ledger.
#
#
# 2015-03-05 wbsch
#   - Now with "I forgot to start/stop this task!" convenience features:
#       "task $id start $x"
#       "task $id stop $x"
#       "task $id done $x"
#     Where $x is the time in minutes you want the entry in your timelog
#     file to be backdated. Note that this is not properly displayed in
#     Taskwarrior itself, but only in your timelog file.
#
#     Note: This will only work on Taskwarrior 2.4.2+ due to a bug in
#           earlier versions. The basic time tracking functionality will
#           work on 2.4.1+.
#
#
# May the Holy Python forgive me for this mess.
#
import calendar
import json
import re

# Debug configuration (default 0, triggered by tw --debug=2)
DEBUG_MODE = 0

tw_debug_level = os.environ.get('TW_DEBUG', '0')
try:
    tw_debug_level = int(tw_debug_level)
except ValueError:
    tw_debug_level = 0

debug_active = DEBUG_MODE == 1 or tw_debug_level >= 2

def get_log_dir():
    cwd = Path.cwd()
    if (cwd / '.git').exists():
        log_dir = cwd / 'logs' / 'debug'
    else:
        log_dir = Path.home() / '.task' / 'logs' / 'debug'
    log_dir.mkdir(parents=True, exist_ok=True)
    return log_dir

if debug_active:
    DEBUG_LOG_DIR = get_log_dir()
    DEBUG_SESSION_ID = datetime.now().strftime("%Y%m%d_%H%M%S")
    try:
        script_name = Path(__file__).stem
    except:
        script_name = Path(sys.argv[0]).stem if sys.argv else "script"
    DEBUG_LOG_FILE = DEBUG_LOG_DIR / f"{script_name}_debug_{DEBUG_SESSION_ID}.log"
    
    def debug_log(message, level=1):
        if debug_active and tw_debug_level >= level:
            timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
            log_line = f"{timestamp} [DEBUG-{level}] {message}\n"
            with open(DEBUG_LOG_FILE, "a") as f:
                f.write(log_line)
            print(f"\033[34m[DEBUG-{level}]\033[0m {message}", file=sys.stderr)
    
    with open(DEBUG_LOG_FILE, "w") as f:
        f.write("=" * 70 + "\n")
        f.write(f"Debug Session - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write(f"Script: {script_name}\n")
        f.write(f"TW_DEBUG Level: {tw_debug_level}\n")
        f.write("=" * 70 + "\n\n")
    debug_log(f"Debug logging initialized: {DEBUG_LOG_FILE}", 1)
else:
    def debug_log(message, level=1):
        pass

# ============================================================================
# Original Code
# ============================================================================

LEDGERFILE = "%s/.task/hooks/timelog/tw.timeclock" % os.getenv('HOME')

if 'TIMELOG' in os.environ:
    LEDGERFILE = os.environ['TIMELOG']


def adjust_date(d, adjust_by):
    if not isinstance(d, datetime):
        d = tw_to_dt(d)
    d -= timedelta(minutes=int(adjust_by))
    return d

def tw_to_dt(s):
    """ Taskwarrior JSON date ---> datetime object. """
    return datetime.strptime(s, "%Y%m%dT%H%M%SZ")

def dt_to_tw(d):
    """ datetime object ---> Taskwarrior JSON date. """
    return d.strftime("%Y%m%dT%H%M%SZ")


if __name__ == '__main__':
    old = json.loads(sys.stdin.readline())
    new = json.loads(sys.stdin.readline())

    annotation_added = ('annotations' in new and not 'annotations' in old) \
                        or \
                     ('annotations' in new and 'annotations' in old and \
                      len(new['annotations']) > len(old['annotations']))


    # task started
    if ('start' in new and not 'start' in old) and annotation_added:
        new['annotations'].sort(key=lambda anno: anno['entry'])
        m = re.match('^[0-9]+$', new['annotations'][-1]['description'])
        if m:
            new['start'] = dt_to_tw(adjust_date(new['start'], int(m.group(0))))
            new['annotations'] = new['annotations'][:-1]
            if not new['annotations']:
                del new['annotations']
            print("Timelog: Started task %s minutes ago." % m.group(0))

            if tw_to_dt(new['start']) < tw_to_dt(new['entry']):
                new['entry'] = new['start']

    # task stopped
    if 'start' in old and not 'start' in new:
        started_utc = tw_to_dt(old['start'])
        started_ts = calendar.timegm(started_utc.timetuple())
        started = datetime.fromtimestamp(started_ts)
        stopped = datetime.now()

        if annotation_added:
            new['annotations'].sort(key=lambda anno: anno['entry'])
            m = re.match('^[0-9]+$', new['annotations'][-1]['description'])
            if m:
                new['annotations'] = new['annotations'][:-1]
                if not new['annotations']:
                    del new['annotations']
                stopped = adjust_date(stopped, m.group(0))
                if stopped < started:
                    print("ERROR: Stop date -%s minutes would be before the start date!" % m.group(0))
                    sys.exit(1)
                print("Timelog: Stopped task %s minutes ago." % m.group(0))

        # time in
        tw_entry = "i " + started.strftime("%Y/%m/%d %H:%M:%S") + " "

        #add project
        tw_entry += new['project'].replace('.', ':') if 'project' in new else "no project"

        # add description
        tw_entry += "  " + new['description']

        # add tags
        tw_entry += "  ; " + ":,".join(new['tags']) + ":," if 'tags' in new else ""

        # add uuid
        tw_entry += " uuid: " + new['uuid'] + "\n"

        # add time out
        tw_entry += "o " + stopped.strftime("%Y/%m/%d %H:%M:%S") + "\n\n"
        with open(LEDGERFILE, "a") as ledger:
            ledger.write(tw_entry.encode("utf-8"))

    print(json.dumps(new))