Feat: time commands

.env.sample

@@ -1,3 +1,4 @@
+GAMCFGDIR=/home/compiler/.config/compiler-admin/gam
 HARVEST_CLIENT_NAME=Client1
 HARVEST_DATA=data/harvest-sample.csv
 TOGGL_DATA=data/toggl-sample.csv

README.md

@@ -10,17 +10,19 @@ Built on top of [GAMADV-XTD3](https://github.com/taers232c/GAMADV-XTD3) and [GYB
 
 ```bash
 $ compiler-admin -h
-usage: compiler-admin [-h] [-v] {info,init,user} ...
+usage: compiler-admin [-h] [-v] {info,init,time,user} ...
 
 positional arguments:
-  {info,init,user}  The command to run
-    info            Print configuration and debugging information.
-    init            Initialize a new admin project. This command should be run once before any others.
-    user            Work with users in the Compiler org.
+  {info,init,time,user}
+                        The command to run
+    info                Print configuration and debugging information.
+    init                Initialize a new admin project. This command should be run once before any others.
+    time                Work with Compiler time entries.
+    user                Work with users in the Compiler org.
 
 options:
-  -h, --help        show this help message and exit
-  -v, --version     show program's version number and exit
+  -h, --help            show this help message and exit
+  -v, --version         show program's version number and exit
 ```
 
 ## Getting started
@@ -54,6 +56,37 @@ The `init` commands follows the steps in the [GAMADV-XTD3 Wiki](https://github.c
 
 Additionally, GYB is used for Gmail backup/restore. See the [GYB Wiki](https://github.com/GAM-team/got-your-back/wiki) for more information.
 
+## Working with time entires
+
+The `time` command provides an interface for working with time entries from Compiler's various systems:
+
+```bash
+$ compiler-admin time -h
+usage: compiler-admin time [-h] {convert} ...
+
+positional arguments:
+  {convert}   The time command to run.
+    convert   Convert a time report from one format into another.
+
+options:
+  -h, --help  show this help message and exit
+```
+
+### Converting an hours report
+
+With a CSV exported from either Harvest or Toggl, use this command to convert to the opposite format:
+
+```bash
+$ compiler-admin time convert -h
+usage: compiler-admin time convert [-h] [--input INPUT] [--output OUTPUT] [--client CLIENT]
+
+options:
+  -h, --help       show this help message and exit
+  --input INPUT    The path to the source data for conversion. Defaults to stdin.
+  --output OUTPUT  The path to the file where converted data should be written. Defaults to stdout.
+  --client CLIENT  The name of the client to use in converted data.
+```
+
 ## Working with users
 
 The following commands are available to work with users in the Compiler domain:

compiler_admin/commands/time/__init__.py

@@ -0,0 +1,15 @@
+from argparse import Namespace
+
+from compiler_admin.commands.time.convert import convert  # noqa: F401
+
+
+def time(args: Namespace, *extra):
+    # try to call the subcommand function directly from global (module) symbols
+    # if the subcommand function was imported above, it should exist in globals()
+    global_env = globals()
+
+    if args.subcommand in global_env:
+        cmd_func = global_env[args.subcommand]
+        cmd_func(args, *extra)
+    else:
+        raise NotImplementedError(f"Unknown time subcommand: {args.subcommand}")

compiler_admin/commands/time/convert.py

@@ -0,0 +1,26 @@
+from argparse import Namespace
+
+import pandas as pd
+
+from compiler_admin import RESULT_SUCCESS
+from compiler_admin.services.harvest import INPUT_COLUMNS as TOGGL_COLUMNS, convert_to_toggl
+from compiler_admin.services.toggl import INPUT_COLUMNS as HARVEST_COLUMNS, convert_to_harvest
+
+
+def _get_source_converter(source):
+    columns = pd.read_csv(source, nrows=0).columns.tolist()
+
+    if set(HARVEST_COLUMNS) <= set(columns):
+        return convert_to_harvest
+    elif set(TOGGL_COLUMNS) <= set(columns):
+        return convert_to_toggl
+    else:
+        raise NotImplementedError("A converter for the given source data does not exist.")
+
+
+def convert(args: Namespace, *extras):
+    converter = _get_source_converter(args.input)
+
+    converter(args.input, args.output, args.client)
+
+    return RESULT_SUCCESS

compiler_admin/commands/user/__init__.py

@@ -10,9 +10,12 @@
 
 
 def user(args: Namespace, *extra):
-    # try to call the subcommand function directly from local symbols
-    # if the subcommand function was imported above, it should exist in locals()
-    if args.subcommand in locals():
-        locals()[args.subcommand](args, *extra)
+    # try to call the subcommand function directly from global (module) symbols
+    # if the subcommand function was imported above, it should exist in globals()
+    global_env = globals()
+
+    if args.subcommand in global_env:
+        cmd_func = global_env[args.subcommand]
+        cmd_func(args, *extra)
     else:
-        raise ValueError(f"Unknown user subcommand: {args.subcommand}")
+        raise NotImplementedError(f"Unknown user subcommand: {args.subcommand}")

compiler_admin/main.py

@@ -4,78 +4,106 @@
 from compiler_admin import __version__ as version
 from compiler_admin.commands.info import info
 from compiler_admin.commands.init import init
+from compiler_admin.commands.time import time
 from compiler_admin.commands.user import user
 from compiler_admin.commands.user.convert import ACCOUNT_TYPE_OU
 
 
+def add_sub_cmd_parser(parser: ArgumentParser, dest="subcommand", help=None):
+    """Helper adds a subparser for the given dest."""
+    return parser.add_subparsers(dest=dest, help=help)
+
+
 def add_sub_cmd(cmd: _SubParsersAction, subcmd, help) -> ArgumentParser:
     """Helper creates a new subcommand parser."""
     return cmd.add_parser(subcmd, help=help)
 
 
-def add_sub_cmd_username(cmd: _SubParsersAction, subcmd, help) -> ArgumentParser:
+def add_sub_cmd_with_username_arg(cmd: _SubParsersAction, subcmd, help) -> ArgumentParser:
     """Helper creates a new subcommand parser with a required username arg."""
-    return add_username_arg(add_sub_cmd(cmd, subcmd, help=help))
-
-
-def add_username_arg(cmd: ArgumentParser) -> ArgumentParser:
-    cmd.add_argument("username", help="A Compiler user account name, sans domain.")
-    return cmd
-
+    sub_cmd = add_sub_cmd(cmd, subcmd, help=help)
+    sub_cmd.add_argument("username", help="A Compiler user account name, sans domain.")
+    return sub_cmd
 
-def main(argv=None):
-    argv = argv if argv is not None else sys.argv[1:]
-    parser = ArgumentParser(prog="compiler-admin")
-
-    # https://stackoverflow.com/a/8521644/812183
-    parser.add_argument(
-        "-v",
-        "--version",
-        action="version",
-        version=f"%(prog)s {version}",
-    )
-
-    cmd_parsers = parser.add_subparsers(dest="command", help="The command to run")
 
+def setup_info_command(cmd_parsers: _SubParsersAction):
     info_cmd = add_sub_cmd(cmd_parsers, "info", help="Print configuration and debugging information.")
     info_cmd.set_defaults(func=info)
 
-    init_cmd = add_sub_cmd_username(
+
+def setup_init_command(cmd_parsers: _SubParsersAction):
+    init_cmd = add_sub_cmd_with_username_arg(
         cmd_parsers, "init", help="Initialize a new admin project. This command should be run once before any others."
     )
     init_cmd.add_argument("--gam", action="store_true", help="If provided, initialize a new GAM project.")
     init_cmd.add_argument("--gyb", action="store_true", help="If provided, initialize a new GYB project.")
     init_cmd.set_defaults(func=init)
 
+
+def setup_time_command(cmd_parsers: _SubParsersAction):
+    time_cmd = add_sub_cmd(cmd_parsers, "time", help="Work with Compiler time entries.")
+    time_cmd.set_defaults(func=time)
+    time_subcmds = add_sub_cmd_parser(time_cmd, help="The time command to run.")
+
+    time_convert = add_sub_cmd(time_subcmds, "convert", help="Convert a time report from one format into another.")
+    time_convert.add_argument(
+        "--input", default=sys.stdin, help="The path to the source data for conversion. Defaults to stdin."
+    )
+    time_convert.add_argument(
+        "--output", default=sys.stdout, help="The path to the file where converted data should be written. Defaults to stdout."
+    )
+    time_convert.add_argument("--client", default=None, help="The name of the client to use in converted data.")
+
+
+def setup_user_command(cmd_parsers: _SubParsersAction):
     user_cmd = add_sub_cmd(cmd_parsers, "user", help="Work with users in the Compiler org.")
     user_cmd.set_defaults(func=user)
-    user_subcmds = user_cmd.add_subparsers(dest="subcommand", help="The user command to run.")
+    user_subcmds = add_sub_cmd_parser(user_cmd, help="The user command to run.")
 
-    user_create = add_sub_cmd_username(user_subcmds, "create", help="Create a new user in the Compiler domain.")
+    user_create = add_sub_cmd_with_username_arg(user_subcmds, "create", help="Create a new user in the Compiler domain.")
     user_create.add_argument("--notify", help="An email address to send the newly created account info.")
 
-    user_convert = add_sub_cmd_username(user_subcmds, "convert", help="Convert a user account to a new type.")
+    user_convert = add_sub_cmd_with_username_arg(user_subcmds, "convert", help="Convert a user account to a new type.")
     user_convert.add_argument("account_type", choices=ACCOUNT_TYPE_OU.keys(), help="Target account type for this conversion.")
 
-    user_delete = add_sub_cmd_username(user_subcmds, "delete", help="Delete a user account.")
+    user_delete = add_sub_cmd_with_username_arg(user_subcmds, "delete", help="Delete a user account.")
     user_delete.add_argument("--force", action="store_true", default=False, help="Don't ask for confirmation before deletion.")
 
-    user_offboard = add_sub_cmd_username(user_subcmds, "offboard", help="Offboard a user account.")
+    user_offboard = add_sub_cmd_with_username_arg(user_subcmds, "offboard", help="Offboard a user account.")
     user_offboard.add_argument("--alias", help="Account to assign username as an alias.")
     user_offboard.add_argument(
         "--force", action="store_true", default=False, help="Don't ask for confirmation before offboarding."
     )
 
-    user_reset = add_sub_cmd_username(
+    user_reset = add_sub_cmd_with_username_arg(
         user_subcmds, "reset-password", help="Reset a user's password to a randomly generated string."
     )
     user_reset.add_argument("--notify", help="An email address to send the newly generated password.")
 
-    add_sub_cmd_username(user_subcmds, "restore", help="Restore an email backup from a prior offboarding.")
+    add_sub_cmd_with_username_arg(user_subcmds, "restore", help="Restore an email backup from a prior offboarding.")
 
-    user_signout = add_sub_cmd_username(user_subcmds, "signout", help="Signs a user out from all active sessions.")
+    user_signout = add_sub_cmd_with_username_arg(user_subcmds, "signout", help="Signs a user out from all active sessions.")
     user_signout.add_argument("--force", action="store_true", default=False, help="Don't ask for confirmation before signout.")
 
+
+def main(argv=None):
+    argv = argv if argv is not None else sys.argv[1:]
+    parser = ArgumentParser(prog="compiler-admin")
+
+    # https://stackoverflow.com/a/8521644/812183
+    parser.add_argument(
+        "-v",
+        "--version",
+        action="version",
+        version=f"%(prog)s {version}",
+    )
+
+    cmd_parsers = add_sub_cmd_parser(parser, dest="command", help="The command to run")
+    setup_info_command(cmd_parsers)
+    setup_init_command(cmd_parsers)
+    setup_time_command(cmd_parsers)
+    setup_user_command(cmd_parsers)
+
     if len(argv) == 0:
         argv = ["info"]
 

compiler_admin/services/files.py

@@ -0,0 +1,25 @@
+import json
+
+import pandas as pd
+
+
+def read_csv(file_path, **kwargs) -> pd.DataFrame:
+    """Read a file path or buffer of CSV data into a pandas.DataFrame."""
+    return pd.read_csv(file_path, **kwargs)
+
+
+def read_json(file_path: str):
+    """Read a file path of JSON data into a python object."""
+    with open(file_path, "r") as f:
+        return json.load(f)
+
+
+def write_csv(file_path, data: pd.DataFrame, columns: list[str] = None):
+    """Write a pandas.DataFrame as CSV to the given path or buffer, with an optional list of columns to write."""
+    data.to_csv(file_path, columns=columns, index=False)
+
+
+def write_json(file_path: str, data):
+    """Write a python object as JSON to the given path."""
+    with open(file_path, "w") as f:
+        json.dump(data, f, indent=2)

compiler_admin/services/harvest.py

@@ -0,0 +1,91 @@
+from datetime import datetime, timedelta
+import os
+import sys
+from typing import TextIO
+
+import pandas as pd
+
+import compiler_admin.services.files as files
+
+# input CSV columns needed for conversion
+INPUT_COLUMNS = ["Date", "Client", "Project", "Notes", "Hours", "First name", "Last name"]
+
+# default output CSV columns
+OUTPUT_COLUMNS = ["Email", "Start date", "Start time", "Duration", "Project", "Task", "Client", "Billable", "Description"]
+
+
+def _calc_start_time(group: pd.DataFrame):
+    """Start time is offset by the previous record's duration, with a default of 0 offset for the first record."""
+    group["Start time"] = group["Start time"] + group["Duration"].shift(fill_value=pd.to_timedelta("00:00:00")).cumsum()
+    return group
+
+
+def _duration_str(duration: timedelta) -> str:
+    """Use total seconds to convert to a datetime and format as a string e.g. 01:30."""
+    return datetime.fromtimestamp(duration.total_seconds()).strftime("%H:%M")
+
+
+def _toggl_client_name():
+    """Gets the value of the TOGGL_CLIENT_NAME env var."""
+    return os.environ.get("TOGGL_CLIENT_NAME")
+
+
+def convert_to_toggl(
+    source_path: str | TextIO = sys.stdin,
+    output_path: str | TextIO = sys.stdout,
+    client_name: str = None,
+    output_cols: list[str] = OUTPUT_COLUMNS,
+):
+    """Convert Harvest formatted entries in source_path to equivalent Toggl formatted entries.
+
+    Args:
+        source_path: The path to a readable CSV file of Harvest time entries; or a readable buffer of the same.
+
+        output_cols (list[str]): A list of column names for the output
+
+        output_path: The path to a CSV file where Toggl time entries will be written; or a writeable buffer for the same.
+
+    Returns:
+        None. Either prints the resulting CSV data or writes to output_path.
+    """
+    if client_name is None:
+        client_name = _toggl_client_name()
+
+    # read CSV file, parsing dates
+    source = files.read_csv(source_path, usecols=INPUT_COLUMNS, parse_dates=["Date"], cache_dates=True)
+
+    # rename columns that can be imported as-is
+    source.rename(columns={"Project": "Task", "Notes": "Description", "Date": "Start date"}, inplace=True)
+
+    # update static calculated columns
+    source["Client"] = client_name
+    source["Project"] = client_name
+    source["Billable"] = "Yes"
+
+    # add the Email column
+    source["Email"] = source["First name"].apply(lambda x: f"{x.lower()}@compiler.la")
+
+    # Convert numeric Hours to timedelta Duration
+    source["Duration"] = source["Hours"].apply(pd.to_timedelta, unit="hours")
+
+    # Default start time to 09:00
+    source["Start time"] = pd.to_timedelta("09:00:00")
+
+    user_days = (
+        source
+        # sort and group by email and date
+        .sort_values(["Email", "Start date"]).groupby(["Email", "Start date"], observed=False)
+        # calculate a start time within each group (excluding the groupby columns)
+        .apply(_calc_start_time, include_groups=False)
+    )
+
+    # convert timedeltas to duration strings
+    user_days["Duration"] = user_days["Duration"].apply(_duration_str)
+    user_days["Start time"] = user_days["Start time"].apply(_duration_str)
+
+    # re-sort by start date/time and user
+    # reset the index to get rid of the group multi index and fold the group columns back down
+    output_data = pd.DataFrame(data=user_days).reset_index()
+    output_data.sort_values(["Start date", "Start time", "Email"], inplace=True)
+
+    files.write_csv(output_path, output_data, output_cols)