feat(toggl): convert to Justworks formatted data

Author: thekaveman

compiler_admin/services/toggl.py

@@ -16,6 +16,8 @@
 
 # default output CSV columns for Harvest
 HARVEST_COLUMNS = ["Date", "Client", "Project", "Task", "Notes", "Hours", "First name", "Last name"]
+# default output CSV columns for Justworks
+JUSTWORKS_COLUMNS = ["First Name", "Last Name", "Work Email", "Start Date", "End Date", "Regular Hours"]
 
 
 @cache
@@ -121,6 +123,52 @@ def convert_to_harvest(
     files.write_csv(output_path, source, columns=output_cols)
 
 
+def convert_to_justworks(
+    source_path: str | TextIO = sys.stdin,
+    output_path: str | TextIO = sys.stdout,
+    output_cols: list[str] = JUSTWORKS_COLUMNS,
+    **kwargs,
+):
+    """Convert Toggl formatted entries in source_path to equivalent Justworks formatted entries.
+
+    Args:
+        source_path: The path to a readable CSV file of Toggl time entries; or a readable buffer of the same.
+
+        output_path: The path to a CSV file where Harvest time entries will be written; or a writeable buffer for the same.
+
+        output_cols (list[str]): A list of column names for the output
+
+    Returns:
+        None. Either prints the resulting CSV data or writes to output_path.
+    """
+    source = _prepare_input(
+        source_path=source_path,
+        column_renames={
+            "Email": "Work Email",
+            "First name": "First Name",
+            "Hours": "Regular Hours",
+            "Last name": "Last Name",
+            "Start date": "Start Date",
+        },
+    )
+
+    # aggregate hours per person per day
+    cols = ["Work Email", "First Name", "Last Name", "Start Date"]
+    people = source.sort_values(cols).groupby(cols, observed=False)
+    people_agg = people.agg({"Regular Hours": "sum"})
+    people_agg.reset_index(inplace=True)
+
+    # aggregate hours per person and rollup to the week (starting on Sunday)
+    cols = ["Work Email", "First Name", "Last Name"]
+    weekly_agg = people_agg.groupby(cols).resample("W", label="left", on="Start Date")
+    weekly_agg = weekly_agg["Regular Hours"].sum().reset_index()
+
+    # calculate the week end date (the following Saturday)
+    weekly_agg["End Date"] = weekly_agg["Start Date"] + pd.Timedelta(days=6)
+
+    files.write_csv(output_path, weekly_agg, columns=output_cols)
+
+
 def download_time_entries(
     start_date: datetime,
     end_date: datetime,

tests/conftest.py

@@ -191,6 +191,11 @@ def harvest_file():
     return "notebooks/data/harvest-sample.csv"
 
 
+@pytest.fixture
+def justworks_file():
+    return "notebooks/data/justworks-sample.csv"
+
+
 @pytest.fixture
 def toggl_file():
     return "notebooks/data/toggl-sample.csv"

tests/services/test_toggl.py

@@ -15,9 +15,11 @@
     _prepare_input,
     _str_timedelta,
     convert_to_harvest,
+    convert_to_justworks,
     download_time_entries,
     TOGGL_COLUMNS,
     HARVEST_COLUMNS,
+    JUSTWORKS_COLUMNS,
     files,
 )
 
@@ -200,6 +202,34 @@ def test_convert_to_harvest_sample(toggl_file, harvest_file, mock_google_user_in
     assert output_df["Client"].eq("Test Client 123").all()
 
 
+def test_convert_to_justworks_mocked(toggl_file, spy_files):
+    convert_to_justworks(toggl_file)
+
+    spy_files.write_csv.assert_called_once()
+    call_args = spy_files.write_csv.call_args
+    assert sys.stdout in call_args[0]
+    assert call_args.kwargs["columns"] == JUSTWORKS_COLUMNS
+
+
+def test_convert_to_justworks_sample(toggl_file, justworks_file):
+    output = None
+
+    with StringIO() as output_data:
+        convert_to_justworks(toggl_file, output_data)
+        output = output_data.getvalue()
+
+    assert output
+    assert isinstance(output, str)
+    assert ",".join(JUSTWORKS_COLUMNS) in output
+
+    order = ["Start Date", "First Name", "Regular Hours"]
+    sample_output_df = pd.read_csv(justworks_file).sort_values(order)
+    output_df = pd.read_csv(StringIO(output)).sort_values(order)
+
+    assert set(output_df.columns.to_list()) <= set(sample_output_df.columns.to_list())
+    assert output_df.shape == sample_output_df.shape
+
+
 @pytest.mark.usefixtures("mock_toggl_api_env", "mock_toggl_detailed_time_entries")
 def test_download_time_entries(toggl_file):
     dt = datetime.now()