Toggl → Timewarrior import extension

toggl_import.py

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

© 2016 Paul Natsuo Kishimoto <[email protected]>
  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 ~/.config/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

- Imports time entries from the last 365 days (Toggl API limitation)
- Imported entries have the following tags:
  - The Toggl entry description
  - If the entry ends with ' uuid:' and then a UUID, this portion is truncated
    and added as a separate 'uuid:<UUID>' tag
  - Any Toggl workspace tags
  - The Toggl project name, with the prefix 'project:'

- The extension respects the Toggle API rate limit of 1 request per second
- Times are converted to UTC
- If :debug is given, each entry is tagged with the Toggl entry ID ('toggl_id:')

API VERSION

This extension uses Toggl Track API v9/v3 (as of 2024)

HISTORY

2024-01
- Update to support new Toggl Track API (v9/v3)
- Changed to use new endpoints for time entries and tags
- Limited to 365-day window due to API restrictions
- Added proper workspace tag support
- Improved timestamp handling
- Updated project and workspace URL handling

2016
- Initial release by Paul Natsuo Kishimoto

"""
from datetime import datetime, timezone, timedelta
from base64 import b64encode
import os
import re
import sys
import time

import requests


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


def get_entries(ws_id, since, projects):
    """Retrieve the detailed report of all tasks using the v3 API."""
    imported = {}

    # Setup initial request parameters
    params = {
        "start_date": since,
        "end_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "page_size": 50,  # Maximum allowed by API
    }

    url = f"https://api.track.toggl.com/reports/api/v3/workspace/{ws_id}/search/time_entries"

    while True:
        result = toggl("POST", url=url, json=params)

        # Process entries
        for entry in result.json():
            if "time_entries" not in entry:
                continue
            item = TogglEntry(entry, projects, tags)

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

        # Check for next page
        next_row = result.headers.get("X-Next-Row-Number")
        if not next_row:
            break

        # Update params for next page
        params["first_row_number"] = int(next_row)

    return imported


def get_tags(ws_id):
    """Retrieve workspace tags."""
    tags = {}
    for tag in toggl("GET", f"workspaces/{ws_id}/tags").json():
        tags[tag["id"]] = tag["name"]

    if DEBUG:
        print("Retrieved %d tags:\n%s" % (len(tags), tags))

    return tags


def get_projects(ws_id):
    """Retrieve project names."""
    projects = {}
    for project in toggl("GET", "workspaces/%d/projects" % int(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.time() - lastTimeCalled[0]
            leftToWait = minInterval - elapsed
            if leftToWait > 0:
                time.sleep(leftToWait)
            ret = func(*args, **kargs)
            lastTimeCalled[0] = time.time()
            return ret

        return rateLimitedFunction

    return decorate


def select_workspace(config):
    """Determine the workspace ID."""
    # List workspaces available through this API key
    response = toggl("GET", "workspaces")
    workspaces = {ws["id"]: ws for ws in response.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[int(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, tags):
        self._data = data
        self._tags = tags

        # Get the first (and typically only) time entry
        self._time_entry = data["time_entries"][0]

        desc, uuid = self.desc_re.match(data.get("description", "")).groups()
        self.description = desc
        self.uuid = uuid

        self.id = self._time_entry["id"]
        self.project = projects.get(data.get("project_id"))

        # Parse ISO format timestamps directly
        self._start = datetime.fromisoformat(self._time_entry["start"])
        self._end = datetime.fromisoformat(self._time_entry["stop"])
        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,
        ]
        # Add any tags
        for tag_id in self._data.get("tag_ids", []):
            if tag_id in self._tags:
                result.append('"%s"' % self._tags[tag_id])
        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)
    tags = get_tags(workspace_id)

    # Use a 365 day window as per API limitations
    start_date = (datetime.now(timezone.utc) - timedelta(days=365)).strftime("%Y-%m-%d")
    all_entries = get_entries(workspace_id, start_date, projects)

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

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