Toggl → Timewarrior import extension

toggl_import.py

#!/usr/bin/env python3
"""Toggl → Timewarrior import extension

© 2016 Paul Natsuo Kishimoto <mail@paul.kishimoto.name>
  Licensed under the GNU GPL v3 or later.

Implements a Timewarrior extension (see
https://taskwarrior.org/docs/timewarrior/index.html) to import data from Toggl
(http://toggl.com).


USAGE

1. Download or clone to ~/.timewarrior/extensions/, make executable.

2. Identify your Toggle API token from https://toggl.com/app/profile, and set
   the Timewarrior configuration variable toggle.api_token:

    $ timew config toggl.api_token ab1c2345defa67b8c9de0f123abc45d6

3. (Optional) Set the workspace ID. If you have only one Toggl workspace,
   toggl_import.py will import from that workspace by default. If you have more
   than one:

   a. Visit https://toggl.com/app/workspaces.
   b. Next to the workspace from which you wish to import, click "… > Settings"
   c. Note the URL: https://toggl.com/app/workspaces/1234567/settings. The
      number 1234567 is the workspace ID.
   d. Set the Timewarrior configuration variable toggle.workspace_id:

    $ timew config toggl.workspace_id 1234567

4. Import:

    $ timew toggl_import :debug
    Wrote 64 entries to /home/username/.timewarrior/data/2015-12.data
    Wrote 114 entries to /home/username/.timewarrior/data/2016-01.data
    Wrote 17 entries to /home/username/.timewarrior/data/2016-02.data
    Wrote 32 entries to /home/username/.timewarrior/data/2016-03.data
    Wrote 3 entries to /home/username/.timewarrior/data/2016-04.data
    Wrote 32 entries to /home/username/.timewarrior/data/2016-05.data
    Wrote 27 entries to /home/username/.timewarrior/data/2016-06.data
    Wrote 46 entries to /home/username/.timewarrior/data/2016-07.data
    Wrote 17 entries to /home/username/.timewarrior/data/2016-08.data


DETAILS

- Imported entries have the following tags:
  - The Toggl entry description. If the entry ends with ' uuid:' and then a
    UUID (such as reported by task <filter> _uuids), then this portion of the
    description is truncated, and a separate tag 'uuid:<UUID>' is added.
  - Any Toggl tags.
  - The Toggl project name, with the prefix 'project:'.

- All Toggl time entries from the creation date of the workspace are imported.

- The extension respects the Toggle API rate limit of 1 request per second per
  IP (https://github.com/toggl/toggl_api_docs#the-api-format). Since the
  Toggl Reports API paginates at 50 time entries per response, the extension
  will take about 2 seconds per every 100 entries imported.

- Times are converted to UTC.

- If :debug is given, each entry is tagged with the Toggl entry ID, with the
  prefix 'toggl_id:'.

"""
from datetime import datetime, timezone
import os
import re
import sys
import time

import requests


DEBUG = False
TOGGL_API_URL = "https://toggl.com/api/v8"
TOGGL_API_TOKEN = None
TOGGL_SESSION_COOKIE = None


def get_entries(ws_id, since, projects):
    """Retrieve the detailed report of all tasks."""

    # Required parameters for the Toggle Reports API
    params = {
        'page': 0,
        'workspace_id': ws_id,
        'since': since,
        }

    # Number of pages of entries to read
    max_pages = sys.maxsize

    # Number of entries to read
    total_count = None

    # Imported entries
    imported = {}

    while params['page'] < max_pages:
        params['page'] += 1  # Page numbering starts at 1
        result = toggl('GET', url='https://toggl.com/reports/api/v2/details',
                       params=params)
        result = result.json()  # Raises an exception on any error

        # Determine the number of pages to retrieve
        if total_count is None:
            # First request. Divide the total_count by the page size to get
            # the number of pages of results
            total_count = result['total_count']
            max_pages = total_count // result['per_page'] + 1
        else:
            assert total_count == result['total_count'], (
                'Number of results changed during import.')

        # Import individual entries
        for entry in result['data']:
            item = TogglEntry(entry, projects)

            # Store in lists by month
            if item.month not in imported:
                imported[item.month] = [item]
            else:
                imported[item.month].append(item)

    return imported


def get_projects(ws_id):
    """Retrieve project names."""
    projects = {}
    for project in toggl('GET', 'workspaces/%d/projects' % ws_id).json():
        projects[project['id']] = project['name']

    if DEBUG:
        print('Retrieved %d projects:\n%s' % (len(projects), projects))

    return projects


def timewarrior_extension_input():
    """Extract the configuration settings."""
    # Copied from timew/ext/totals.py
    header = True
    config = dict()
    body = ''
    for line in sys.stdin:
        if header:
            if line == '\n':
                header = False
            else:
                fields = line.strip().split(': ', 2)
                if len(fields) == 2:
                    config[fields[0]] = fields[1]
                else:
                    config[fields[0]] = ''
        else:
            body += line

    return config, body


def ratelimited(maxPerSecond):
    """Decorator for a rate-limited function.

    Source: https://gist.github.com/gregburek/1441055
    """
    minInterval = 1.0 / float(maxPerSecond)

    def decorate(func):
        lastTimeCalled = [0.0]

        def rateLimitedFunction(*args, **kargs):
            elapsed = time.clock() - lastTimeCalled[0]
            leftToWait = minInterval - elapsed
            if leftToWait > 0:
                time.sleep(leftToWait)
            ret = func(*args, **kargs)
            lastTimeCalled[0] = time.clock()
            return ret
        return rateLimitedFunction
    return decorate


def select_workspace(config):
    """Determine the workspace ID."""
    # List workspaces available through this API key
    workspaces = {ws['id']: ws for ws in toggl('GET', 'workspaces').json()}

    ws_id = config.get('toggl.workspace_id', None)
    if ws_id is None:
        # No workspace IP supplied in configuration
        assert len(workspaces) == 1, ('Cannot determine which workspace to '
                                      'import from.')
        ws_id, ws = list(workspaces.items())[0]
    else:
        # Raises IndexError if the configuration is incorrect
        ws = workspaces[ws_id]

    return ws_id, ws


@ratelimited(1)
def toggl(method, path=None, **kwargs):
    """Make a request to the TOGGL API.

    Heavily modified from:
    https://github.com/drobertadams/toggl-cli/blob/master/toggl.py
    """
    if path is not None:
        kwargs['url'] = TOGGL_API_URL + '/' + path

    # Use the session cookie, if it exists
    kwargs['cookies'] = TOGGL_SESSION_COOKIE
    kwargs['auth'] = requests.auth.HTTPBasicAuth(TOGGL_API_TOKEN, 'api_token')

    # Set the user_agent
    if 'params' not in kwargs:
        kwargs['params'] = dict()
    kwargs['params'].update({
        'user_agent': 'https://github.com/khaeru/timewarrior_toggl',
        })

    return requests.request(method, **kwargs)


class TogglEntry:
    """A Toggle time tracking entity.

    This class mostly provides for one-way translation: the constructor
    understands JSON data from the Toggl API describing a time entry; and the
    string representation of the object is a line suitable for a Timewarrior
    data file.

    """
    desc_re = re.compile('(.*?)(?: uuid:([0-9a-f-]{36}))?$')
    time_map = str.maketrans({':': None})
    time_fmt_in = '%Y-%m-%dT%H%M%S%z'
    time_fmt_out = '%Y%m%dT%H%M%SZ'

    def __init__(self, data, projects):
        self._data = data

        desc, uuid = self.desc_re.match(data['description']).groups()
        self.description = desc
        self.uuid = uuid

        self.id = data['id']
        self.project = projects.get(data['pid'])

        # Translate times into UTC
        # NB timewarrior currently (1.0.0 beta) does not understand ISO
        # formatted datetimes, or those with non-'Z' time zone specifiers,
        # in the data files
        self._start = datetime.strptime(data['start'].translate(self.time_map),
                                        self.time_fmt_in)
        self._end = datetime.strptime(data['end'].translate(self.time_map),
                                      self.time_fmt_in)
        self.month = self._start.strftime('%Y-%m')

    def __str__(self):
        result = [
            'inc',
            self._start.astimezone(timezone.utc).strftime(self.time_fmt_out),
            '-',
            self._end.astimezone(timezone.utc).strftime(self.time_fmt_out),
            '#',
            '"%s"' % self.description,
            ]
        result.extend(['"%s"' % tag for tag in self._data['tags']])
        if self.project is not None:
            result.append('"project:%s"' % self.project)
        if self.uuid is not None:
            result.append('"uuid:%s"' % self.uuid)
        if DEBUG:
            result.append('toggl_id:%s' % self.id)
        return ' '.join(result) + '\n'


def write_entries(month, entries):
    line_re = re.compile('toggl_id:([0-9]+)')

    if DEBUG:
        print(month, '\n'.join(map(str, entries)), sep='\n\n', end='\n\n')

    # Filename for output
    fn = os.path.join(configuration['temp.db'], 'data', '%s.data' % month)

    with open(fn, 'a+') as f:
        f.seek(0)

        # Read through file to determine Toggle entries that have already
        # been imported
        existing_ids = set()
        for line in f:
            try:
                existing_ids.add(int(line_re.search(line).groups()[0]))
            except AttributeError:
                continue

        if DEBUG:
            print('Month %s has existing ids: %s' % (month, existing_ids))

        written = 0
        skipped = 0
        for e in entries:
            if e.id in existing_ids:
                skipped += 1
            else:
                f.write(str(e))
                written += 1

        print('Wrote %d, skipped %d existing entries in %s' %
              (written, skipped, fn))


if __name__ == '__main__':
    configuration, _ = timewarrior_extension_input()

    DEBUG = configuration.get('debug') in ['on', 1, '1', 'yes', 'y', 'true']

    # Open the toggl session
    TOGGL_API_TOKEN = configuration['toggl.api_token']
    TOGGL_SESSION_COOKIE = toggl('POST', 'sessions').cookies

    workspace_id, workspace = select_workspace(configuration)

    projects = get_projects(workspace_id)

    # The key 'at' of the current workspace gives its creation time;
    # use this as the start date for requests
    all_entries = get_entries(workspace_id, workspace['at'].split('T')[0],
                              projects)

    for month, entries in sorted(all_entries.items()):
        write_entries(month, entries)

    # Close the session
    toggl('DELETE', 'sessions')

I added casting to int of workspace ID to get it to work. Maybe Toggl changed their API a bit? My changes:

Line 135: for project in toggl('GET', 'workspaces/%d/projects' % int(ws_id)).json():
Line 201: ws = workspaces[int(ws_id)]

Super ugly fix, but hey, it works :)