Gets a report of hours balance and remaining vacation days at a specific date from TimeTac, for yearly accounting

get_balances_by_date.py

import argparse
import csv
import os
import requests
import sys
from decimal import Decimal
from typing import Any

from dotenv import load_dotenv


class TimeTacClient:
  def __init__(
    self,
    account: str,
    client_id: str,
    client_secret: str,
    api_version: str = "v5",
  ):
    self.account = account
    self.base = f"https://api.timetac.com/{account}/{api_version}"
    self.session = requests.Session()
    self.session.headers.update({"Accept": "application/json"})

    self.authenticate(client_id, client_secret)

  def authenticate(self, client_id: str, client_secret: str) -> None:
    url = f"https://api.timetac.com/{self.account}/auth/oauth2/token"

    response = self.session.post(
      url,
      data={
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
      },
      headers={"Content-Type": "application/x-www-form-urlencoded"},
      timeout=30,
    )
    response.raise_for_status()

    token = response.json()["access_token"]
    self.session.headers.update({"Authorization": f"Bearer {token}"})

  def read(self, resource: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    url = f"{self.base}/{resource}/read"

    response = self.session.get(url, params=params or {}, timeout=60)
    response.raise_for_status()

    data = response.json()
    if data.get("Success") is False:
      raise RuntimeError(data)

    return data

  def read_all(
    self,
    resource: str,
    params: dict[str, Any] | None = None,
    limit: int = 1000,
  ) -> list[dict[str, Any]]:
    rows = []
    offset = 0

    while True:
      page_params = {
        **(params or {}),
        "_offset": offset,
        "_limit": limit,
      }

      data = self.read(resource, page_params)
      page = data.get("Results", [])
      rows.extend(page)

      if len(page) < limit:
        break

      offset += limit

    return rows


def decimal_or_none(value: Any) -> Decimal | None:
  if value in (None, ""):
    return None
  return Decimal(str(value))


def get_users(client: TimeTacClient) -> list[dict[str, Any]]:
  return client.read_all(
    "users",
    {
      "_fields": "id,firstname,lastname,fullname,email_address,active",
      "_order_by": "lastname",
    },
  )


def get_archive_history(client: TimeTacClient) -> list[dict[str, Any]]:
  return client.read_all(
    "usersArchiveHistory",
    {
      "_fields": "user_id,active,valid_from,valid_to",
    },
  )


def was_user_active_on_date(
  user_id: int,
  target_date: str,
  archive_history: list[dict[str, Any]],
) -> bool:
  """Check if user was active (not archived) on target_date."""
  # Filter archive history for this user
  user_history = [h for h in archive_history if int(h.get("user_id")) == user_id]
  
  if not user_history:
    # No archive history means user was never archived
    return True
  
  # Sort by valid_from to find the relevant entry
  user_history.sort(key=lambda h: h.get("valid_from", ""))
  
  # Find the entry that covers target_date
  for entry in user_history:
    valid_from = entry.get("valid_from", "")
    valid_to = entry.get("valid_to") or "9999-12-31"  # Treat null as "still active"
    
    if valid_from <= target_date <= valid_to:
      # Found the covering entry
      return entry.get("active") is True or entry.get("active") == 1
  
  # If no entry covers the date, assume active
  return True


def get_holiday_balance_on_date(
  client: TimeTacClient,
  user_id: int,
  target_date: str,
) -> dict[str, Any] | None:
  data = client.read(
    "timesheetAccountings",
    {
      "_fields": (
        "user_id,date,holiday,holiday_adjustments,"
        "holiday_balance,tmp_holiday_balance,working_time_total_balance"
      ),
      "user_id": user_id,
      "date": target_date,
      "_limit": 1,
    },
  )

  rows = data.get("Results", [])
  if not rows:
    return None

  return rows[0]


def make_report(client: TimeTacClient, target_date: str) -> list[dict[str, Any]]:
  rows = []
  archive_history = get_archive_history(client)

  for user in get_users(client):
    user_id = int(user["id"])
    
    # Include user if they were active on target_date
    if not was_user_active_on_date(user_id, target_date, archive_history):
      continue

    accounting = get_holiday_balance_on_date(client, user_id, target_date)

    holiday_balance = None
    tmp_holiday_balance = None
    hours_balance = None

    if accounting:
      holiday_balance = decimal_or_none(accounting.get("holiday_balance"))
      tmp_holiday_balance = decimal_or_none(accounting.get("tmp_holiday_balance"))
      hours_balance = decimal_or_none(accounting.get("working_time_total_balance"))

    remaining = holiday_balance
    if remaining is None:
      remaining = tmp_holiday_balance

    rows.append(
      {
        "user_id": user_id,
        "name": (
          user.get("fullname")
          or f"{user.get('firstname', '')} {user.get('lastname', '')}".strip()
        ),
        "email": user.get("email_address"),
        "target_date": target_date,
        "remaining_vacation_days": str(remaining) if remaining is not None else "",
        "hours_balance": str(hours_balance) if hours_balance is not None else "",
      }
    )

  rows.sort(key=lambda r: r["user_id"])
  return rows


def write_csv(path: str, rows: list[dict[str, Any]]) -> None:
  if not rows:
    return

  with open(path, "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
    writer.writeheader()
    writer.writerows(rows)


def print_report(rows: list[dict[str, Any]]) -> None:
  if not rows:
    print("No results found")
    return

  # Get all fieldnames
  fieldnames = list(rows[0].keys())

  # Calculate column widths
  col_widths = {}
  for field in fieldnames:
    col_widths[field] = max(
      len(field),
      max((len(str(row.get(field, ""))) for row in rows), default=0)
    )

  # Print header
  header = " | ".join(
    field.ljust(col_widths[field]) for field in fieldnames
  )
  print(header)
  print("-" * len(header))

  # Print rows
  for row in rows:
    print(" | ".join(
      str(row.get(field, "")).ljust(col_widths[field]) for field in fieldnames
    ))


if __name__ == "__main__":
  load_dotenv()

  parser = argparse.ArgumentParser(
    description="Get vacation balance for all users on a target date"
  )
  parser.add_argument(
    "target_date",
    help="Target date in YYYY-MM-DD format",
  )
  parser.add_argument(
    "-o", "--output",
    dest="output_filename",
    default=None,
    help="Output CSV filename (if not specified, prints to stdout)",
  )
  args = parser.parse_args()

  client = TimeTacClient(
    account=os.environ["TIMETAC_ACCOUNT"],
    client_id=os.environ["TIMETAC_CLIENT_ID"],
    client_secret=os.environ["TIMETAC_CLIENT_SECRET"],
  )

  report = make_report(client, args.target_date)

  if args.output_filename:
    write_csv(args.output_filename, report)
  else:
    print_report(report)