Harvest rounder

.envrc

@@ -1,3 +1,3 @@
+watch_file shell.nix
+source_env_if_exists .envrc.local
 use flake
-watch_file .envrc.local shell.nix
-[[ -f .envrc.local ]] && source_env .envrc.local

.envrc.local-template

@@ -2,8 +2,8 @@
 export HARVEST_ACCOUNT_ID=<TODO>
 export HARVEST_BEARER_TOKEN=<TODO>
 
-# Used to filter exports
-export HARVEST_USER="<YOUR NAME>"
+# Used to filter exports (picks the current user by default)
+#export HARVEST_USER="<YOUR NAME>"
 
 # optional: sevdesk credentials
 #export SEVDESK_API_TOKEN=<TODO>

README.md

@@ -81,6 +81,24 @@ $ working-days-calculator report.csv
 Working days: 171 from 2022-01-12 00:00:00 to 2022-12-29 00:00:00
 ```
 
+## Harvest Rounder
+
+Round Harvest time entries up to the nearest increment (default: 15 minutes). This tool helps ensure your time entries are rounded consistently for billing purposes.
+
+### Usage
+
+By default, rounds entries for the authenticated user from the past 4 weeks:
+
+```console
+harvest-rounder --dry-run
+```
+
+Apply rounding (will prompt for confirmation):
+
+```console
+harvest-rounder
+```
+
 ## Kimai Usage/Examples
 
 Exports the last month timesheets of user Jon for client Bob  

flake.nix

@@ -31,6 +31,7 @@
           };
           packages = {
             harvest-exporter = pkgs.callPackage ./harvest-exporter.nix { };
+            harvest-rounder = pkgs.callPackage ./harvest-rounder.nix { };
 
             wise-exporter = pkgs.callPackage ./wise-exporter.nix { };
             sevdesk-api = pkgs.python3.pkgs.callPackage ./sevdesk-api { };
@@ -70,6 +71,7 @@
                   "harvest"
                   "harvest_exporter"
                   "harvest_report"
+                  "harvest_rounder"
                   "rest"
                   "kimai"
                   "kimai_exporter"

harvest-rounder.nix

@@ -0,0 +1,30 @@
+{
+  pkgs ? import <nixpkgs> { },
+  lib ? pkgs.lib,
+}:
+pkgs.python3.pkgs.buildPythonApplication {
+  pname = "harvest-rounder";
+  version = "0.0.1";
+  src = lib.fileset.toSource {
+    root = ./.;
+    fileset = lib.fileset.unions [
+      ./pyproject.toml
+      ./README.md
+      ./harvest
+      ./harvest_exporter
+      ./harvest_rounder
+      ./kimai
+      ./kimai_exporter
+      ./rest
+    ];
+  };
+
+  pyproject = true;
+  build-system = [ pkgs.python3.pkgs.hatchling ];
+
+  doCheck = false;
+
+  # Rich is a dependency of the shared pyproject.toml even though
+  # harvest-rounder doesn't use it directly
+  dependencies = [ pkgs.python3.pkgs.rich ];
+}

harvest/__init__.py

@@ -5,6 +5,24 @@
 from rest import http_request
 
 
+def get_current_user(account_id: str, access_token: str) -> str:
+    """Get the name of the currently authenticated user.
+
+    Args:
+        account_id: Harvest account ID
+        access_token: Harvest bearer token
+
+    Returns:
+        The full name of the current user
+    """
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Harvest-Account-id": account_id,
+    }
+    resp = http_request("https://api.harvestapp.com/v2/users/me", headers=headers)
+    return f"{resp['first_name']} {resp['last_name']}"
+
+
 def get_time_entries(
     account_id: str, access_token: str, from_date: int, to_date: int
 ) -> list[dict[str, Any]]:

harvest_exporter/cli.py

@@ -7,7 +7,7 @@
 from datetime import date, datetime, timedelta
 from fractions import Fraction
 
-from harvest import get_time_entries
+from harvest import get_current_user, get_time_entries
 
 from . import Task, aggregate_time_entries, export
 
@@ -46,7 +46,12 @@ def parse_args() -> argparse.Namespace:
         "--user",
         type=str,
         default=os.environ.get("HARVEST_USER"),
-        help="user to filter for (env: HARVEST_USER)",
+        help="Filter by user name (env: HARVEST_USER). Defaults to the authenticated user.",
+    )
+    parser.add_argument(
+        "--all-users",
+        action="store_true",
+        help="Process entries for all users instead of just the authenticated user",
     )
     parser.add_argument(
         "--start",
@@ -142,6 +147,20 @@ def exclude_task(task: Task, args: argparse.Namespace) -> bool:
 
 def main() -> None:
     args = parse_args()
+
+    # Determine which user to filter by:
+    # 1. --all-users: no filtering
+    # 2. --user: use the specified user
+    # 3. default: auto-discover the authenticated user
+    filter_user = None
+    if not args.all_users:
+        if args.user:
+            filter_user = args.user
+        else:
+            filter_user = get_current_user(
+                args.harvest_account_id, args.harvest_bearer_token
+            )
+
     entries = get_time_entries(
         args.harvest_account_id, args.harvest_bearer_token, args.start, args.end
     )
@@ -152,15 +171,15 @@ def main() -> None:
 
     users = aggregate_time_entries(entries, args.hourly_rate, agency_rate)
 
-    if args.user:
-        for_user = users.get(args.user)
+    if filter_user:
+        for_user = users.get(filter_user)
         if not for_user:
             print(
-                f"user {args.user} not found in time range, found {', '.join(users.keys())}",
+                f"user {filter_user} not found in time range, found {', '.join(users.keys())}",
                 file=sys.stderr,
             )
             sys.exit(1)
-        users = {args.user: for_user}
+        users = {filter_user: for_user}
 
     for user in users.values():
         for client in user.clients.values():

harvest_rounder/__init__.py

@@ -0,0 +1,154 @@
+"""Round Harvest time entries to the nearest increment (default: 15 minutes)."""
+
+from dataclasses import dataclass
+from fractions import Fraction
+from typing import Any
+
+from rest import http_request
+
+
+@dataclass
+class TimeEntry:
+    """Represents a Harvest time entry with rounding information."""
+
+    id: int
+    date: str
+    hours: Fraction
+    rounded_hours: Fraction
+    notes: str
+    project: str
+    task: str
+    client: str
+    user: str
+
+    @property
+    def needs_rounding(self) -> bool:
+        """Check if this entry needs to be rounded."""
+        return self.hours != self.rounded_hours
+
+    @property
+    def difference(self) -> Fraction:
+        """Return the difference between rounded and original hours."""
+        return self.rounded_hours - self.hours
+
+
+def round_to_increment(hours: Fraction, increment_minutes: int = 15) -> Fraction:
+    """Round hours up to the next increment.
+
+    Args:
+        hours: The number of hours as a Fraction
+        increment_minutes: The increment in minutes (default: 15)
+
+    Returns:
+        Hours rounded up to the next increment as a Fraction
+    """
+    # Convert increment from minutes to hours as a fraction
+    increment_hours = Fraction(increment_minutes, 60)
+
+    # If hours is zero, return zero
+    if hours == 0:
+        return Fraction(0)
+
+    # Calculate how many increments fit into the hours
+    # We use ceiling division to round up
+    increments = hours / increment_hours
+
+    # If it's already an exact multiple, return as-is
+    if increments.denominator == 1:
+        return hours
+
+    # Otherwise, round up to next increment
+    rounded_increments = int(increments) + 1
+    return increment_hours * rounded_increments
+
+
+def parse_time_entry(entry: dict[str, Any], increment_minutes: int = 15) -> TimeEntry:
+    """Parse a Harvest API time entry into a TimeEntry object.
+
+    Args:
+        entry: Raw time entry from the Harvest API
+        increment_minutes: The increment in minutes for rounding
+
+    Returns:
+        A TimeEntry object with original and rounded hours
+    """
+    hours = Fraction(entry["hours"]).limit_denominator(1000)
+    rounded_hours = round_to_increment(hours, increment_minutes)
+
+    return TimeEntry(
+        id=entry["id"],
+        date=entry["spent_date"],
+        hours=hours,
+        rounded_hours=rounded_hours,
+        notes=entry.get("notes") or "",
+        project=entry["project"]["name"],
+        task=entry["task"]["name"],
+        client=entry["client"]["name"],
+        user=entry["user"]["name"],
+    )
+
+
+def get_time_entries(
+    account_id: str,
+    access_token: str,
+    from_date: int,
+    to_date: int,
+    increment_minutes: int = 15,
+) -> list[TimeEntry]:
+    """Fetch time entries from Harvest and parse them.
+
+    Args:
+        account_id: Harvest account ID
+        access_token: Harvest bearer token
+        from_date: Start date as YYYYMMDD integer
+        to_date: End date as YYYYMMDD integer
+        increment_minutes: The increment in minutes for rounding
+
+    Returns:
+        List of TimeEntry objects
+    """
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Harvest-Account-id": account_id,
+    }
+    url = f"https://api.harvestapp.com/v2/time_entries?from={from_date}&to={to_date}"
+    entries: list[TimeEntry] = []
+    while url is not None:
+        resp = http_request(url, headers=headers)
+        entries.extend(
+            parse_time_entry(entry, increment_minutes) for entry in resp["time_entries"]
+        )
+        url = resp["links"]["next"]
+    return entries
+
+
+def update_time_entry(
+    account_id: str,
+    access_token: str,
+    entry_id: int,
+    hours: Fraction,
+) -> dict[str, Any]:
+    """Update a time entry's hours in Harvest.
+
+    Args:
+        account_id: Harvest account ID
+        access_token: Harvest bearer token
+        entry_id: The ID of the time entry to update
+        hours: The new hours value
+
+    Returns:
+        The updated time entry from the API
+    """
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Harvest-Account-id": account_id,
+        "Content-Type": "application/json",
+    }
+    url = f"https://api.harvestapp.com/v2/time_entries/{entry_id}"
+
+    return http_request(
+        url,
+        method="PATCH",
+        headers=headers,
+        data={"hours": float(hours)},
+    )