ねことくまとへび

2021年最初のプログラムを書きました。100行にも満たないプログラムですが、仕事でやっている言語を離れて好きな言語を使える喜びに満ちた、とても楽しい時間を過ごすことができました。

私は作業時間のトラッキングにToggl Trackを利用しており、各エントリのClientにその時間が消費、投資、浪費のどれに当たるかを指定しています。より厳密には、消費、投資、浪費にぶら下げる形で各Projectを作り、それをエントリに割り当てています。

今日作成したのは、一日の時間の使い方を集計するプログラムです。入力が20210101であれば、2021年1月1日の0時から2日の0時までの24時間を対象に、Clientごとの合計時間を集計します。日をまたいだエントリについては、前日分、翌日分を破棄し、当日24時間の中に入っている部分のみを集計します。

もともとAPIの戻り値にはdurationという作業時間の項目があるのですが、今回は当該エントリの開始時刻から次のエントリの開始時刻までを作業時間としています。アプリの操作やサーバー同期の都合上、どうしても終了→即座に次のエントリ開始というわけには行かず、数秒の空白の時間帯が発生してしまうからです。

from itertools import groupby
from datetime import datetime, timedelta, timezone
from functools import cache
import sys
import time
import requests

API_TOKEN = '<YOUR_API_TOKEN>'
END_POINT = 'https://api.track.toggl.com/api/v8/'
WAIT = 1

def main(target):
    target_date = datetime.strptime(target, '%Y%m%d').replace(tzinfo=timezone(timedelta(hours=9), 'JST'))
    start_date = target_date + timedelta(days=-1)
    end_date = target_date + timedelta(days=1)

    params = {
        "start_date": start_date.isoformat(timespec='seconds'),
        "end_date": end_date.isoformat(timespec='seconds'),
    }
    response = requests.get(END_POINT + 'time_entries', params, auth=(API_TOKEN, 'api_token')).json()

    entries = list(map(
        lambda x: {
            'pid': x['pid'],
            'start': max(datetime.fromisoformat(x['start']), target_date),
        },
        filter(lambda x: datetime.fromisoformat(x['stop']) > target_date, response)
    ))

    durations = [(
        fetch_project(start['pid'])[0],
        stop['start'] - start['start']
    ) for start, stop in zip(entries, entries[1:] + [{'pid': 0, 'start': end_date}])]

    result = [(
        k,
        sum([duration[1] for duration in v], timedelta())
    ) for k, v in groupby(sorted(durations, key=lambda x: x[0]), lambda y: y[0])]

    for k, v in result:
        print(k, format_timedelta(v))

@cache
def fetch_project(pid):
    time.sleep(WAIT)
    response = requests.get(END_POINT + 'projects/' + str(pid), auth=(API_TOKEN, 'api_token')).json()
    return fetch_client(response['data']['cid']), response['data']['name']

@cache
def fetch_client(cid):
    time.sleep(WAIT)
    response = requests.get(END_POINT + 'clients/' + str(cid), auth=(API_TOKEN, 'api_token')).json()
    return response['data']['name']

def format_timedelta(td):
    hours, remainder = divmod(td.seconds, 3600)
    minutes, seconds = divmod(remainder, 60)
    return f"{hours:02}:{minutes:02}:{seconds:02}"

if __name__ == '__main__':
    main(sys.argv[1])