time_tabulation.py

import argparse
import csv
import dataclasses
import datetime
import os
import sqlite3
import sys
import time

from typing import Iterator

from dateutil.parser import parse as parse_date
import requests


API_KEY = os.environ['RESCUETIME_API_KEY']

COMMANDS = {}


def command(fn):
    COMMANDS[fn.__name__] = fn


@dataclasses.dataclass
class SummaryItem:
    date: datetime.date
    duration: datetime.timedelta
    activity: str
    category: str
    productivity: int


def get_summary_data(start_date, end_date) -> Iterator[SummaryItem]:
    results = requests.get(
        'https://www.rescuetime.com/anapi/data',
        headers={
            'Accept': 'application/json',
        },
        params={
            'key': API_KEY,
            'perspective': 'interval',
            'resolution_time': 'day',
            'restrict_begin': start_date.strftime('%Y-%m-%d'),
            'restrict_end': end_date.strftime('%Y-%m-%d'),
        }
    ).json()['rows']

    for row in results:
        yield SummaryItem(
            parse_date(row[0]).date(),
            datetime.timedelta(seconds=row[1]),
            row[3],
            row[4],
            row[5]
        )


@command
def gather_data(extra):
    parser = argparse.ArgumentParser()
    parser.add_argument('start_date', type=parse_date)
    parser.add_argument('end_date', type=parse_date)
    args = parser.parse_args(extra)

    start_date = args.start_date
    end_date = args.end_date

    connection = sqlite3.connect('time.db')
    cursor = connection.cursor()

    cursor.execute('''
        CREATE TABLE time (
            date integer,
            duration integer,
            activity string,
            category string,
            productivity int
        )
    ''')

    for result in get_summary_data(start_date, end_date):
        cursor.execute(
            '''
                INSERT INTO time (
                    date, duration, activity, category, productivity
                )
                VALUES (?, ?, ?, ?, ?)
            ''', (
                int(time.mktime(result.date.timetuple())),
                result.duration.total_seconds(),
                result.activity,
                result.category,
                result.productivity
            )
        )

    connection.commit()


@command
def summarize(extra):
    parser = argparse.ArgumentParser()
    parser.add_argument('--min-day-hours', type=int, default=6)
    args = parser.parse_args(extra)

    connection = sqlite3.connect('time.db')
    cursor = connection.cursor()

    min_duration_hours = args.min_day_hours

    cursor.execute('''
        SELECT
            COUNT(*)
        FROM (
            SELECT
                date,
                sum(duration) as total_duration
            FROM time
            GROUP BY date
        ) WHERE total_duration > (60 * 60 * ?)
    ''', (
        min_duration_hours,
    ))
    day_count = cursor.fetchall()[0][0]

    cursor.execute('''
        SELECT
            category,
            1.0 * SUM(sum_duration) / ? / (60 * 60)
        FROM
            (
                SELECT
                    time.date,
                    time.category,
                    sum(duration) as sum_duration
                FROM time
                INNER JOIN (
                    SELECT
                        date,
                        sum(duration) as total_duration
                    FROM time
                    GROUP BY date
                ) AS allowed_dates ON (
                    allowed_dates.date = time.date
                    /* Count only days having more than an hour recorded */
                    AND allowed_dates.total_duration > (60 * 60 * ?)
                )
                GROUP BY time.date, time.category
            ) AS per_date
        GROUP BY category
        ORDER BY sum_duration DESC
    ''', (
        day_count,
        min_duration_hours,
    ))

    writer = csv.writer(sys.stdout)

    for result in cursor.fetchall():
        writer.writerow(result)


@command
def detail(extra):
    parser = argparse.ArgumentParser()
    parser.add_argument('--min-day-hours', type=int, default=6)
    args = parser.parse_args(extra)

    connection = sqlite3.connect('time.db')
    cursor = connection.cursor()

    min_duration_hours = args.min_day_hours

    cursor.execute('''
        SELECT
            COUNT(*)
        FROM (
            SELECT
                date,
                sum(duration) as total_duration
            FROM time
            GROUP BY date
        ) WHERE total_duration > (60 * 60 * ?)
    ''', (
        min_duration_hours,
    ))
    day_count = cursor.fetchall()[0][0]

    cursor.execute('''
        SELECT
            category,
            activity,
            1.0 * SUM(sum_duration) / ? / (60 * 60)
        FROM
            (
                SELECT
                    time.date,
                    time.category,
                    time.activity,
                    sum(duration) as sum_duration
                FROM time
                INNER JOIN (
                    SELECT
                        date,
                        sum(duration) as total_duration
                    FROM time
                    GROUP BY date
                ) AS allowed_dates ON (
                    allowed_dates.date = time.date
                    /* Count only days having more than an hour recorded */
                    AND allowed_dates.total_duration > (60 * 60 * ?)
                )
                GROUP BY time.date, time.category, time.activity
            ) AS per_date
        GROUP BY category, activity
        ORDER BY sum_duration DESC
    ''', (
        day_count,
        min_duration_hours,
    ))

    writer = csv.writer(sys.stdout)

    for result in cursor.fetchall():
        writer.writerow(result)


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('command', choices=COMMANDS.keys())
    args, extra = parser.parse_known_args()

    COMMANDS[args.command](extra)