#!/usr/bin/env python3 """ One-shot import of Rian's Toggl Detailed Report CSV. Mirrors what the /import page would do, but: - Reads the CSV from .planning/Misc - Attributes the batch to a specific user (Rian) via imported_by - Sets source='toggl' on the batch Run from /srv/apps/work: python3 scripts/import_toggl_csv.py """ from __future__ import annotations import csv import datetime as dt import json import os import re import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent CSV_DIR = PROJECT_ROOT / ".planning/Misc" ORG_SLUG = "bowden-works" BATCH_NAME = "Rian's 2026 Toggl (Jan–May 18)" IMPORTER_EMAIL = "rian@rian.ca" # Load .env for line in (PROJECT_ROOT / ".env").read_text().splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip()) SUPABASE_URL = os.environ["NEXT_PUBLIC_SUPABASE_URL"].rstrip("/") SECRET_KEY = os.environ["SUPABASE_SECRET_KEY"] def sb(method, path, *, body=None, params=None, prefer=None): url = f"{SUPABASE_URL}/rest/v1{path}" if params: url += "?" + urllib.parse.urlencode(params) headers = { "apikey": SECRET_KEY, "Authorization": f"Bearer {SECRET_KEY}", "Content-Type": "application/json", } if prefer: headers["Prefer"] = prefer data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req) as resp: raw = resp.read() return json.loads(raw) if raw else None except urllib.error.HTTPError as e: msg = e.read().decode(errors="replace") raise RuntimeError(f"{method} {path} failed: {e.code} {msg[:500]}") def parse_dur_to_seconds(s: str) -> int | None: s = (s or "").strip() if not s: return None m = re.match(r"^(\d+):(\d{2}):(\d{2})$", s) if m: return int(m[1]) * 3600 + int(m[2]) * 60 + int(m[3]) m = re.match(r"^(\d+):(\d{2})$", s) if m: return int(m[1]) * 3600 + int(m[2]) * 60 try: return int(round(float(s) * 3600)) except ValueError: return None def split_client_project(raw: str): sep = " : " i = raw.find(sep) if i == -1: return (None, raw.strip()) return (raw[:i].strip() or None, raw[i + len(sep):].strip()) def main(): csv_files = sorted(CSV_DIR.glob("TogglTrack_Report_*.csv")) if not csv_files: print(f"No TogglTrack_Report_*.csv found in {CSV_DIR}", file=sys.stderr) sys.exit(1) csv_path = csv_files[-1] print(f"Using {csv_path.name}") # Resolve org + importer + base rate org = sb("GET", "/organizations", params={ "slug": f"eq.{ORG_SLUG}", "select": "id,default_hourly_rate_usd", }) if not org: raise RuntimeError("org not found") org_id = org[0]["id"] base_rate = float(org[0]["default_hourly_rate_usd"]) print(f" org_id={org_id} base_rate=${base_rate}/hr") importer = sb("GET", "/profiles", params={ "email": f"eq.{IMPORTER_EMAIL}", "select": "id", }) if not importer: raise RuntimeError(f"importer profile not found: {IMPORTER_EMAIL}") importer_id = importer[0]["id"] print(f" importer_id={importer_id} ({IMPORTER_EMAIL})") # Team member lookup keyed by lowercased email tm_rows = sb("GET", "/team_members", params={ "org_id": f"eq.{org_id}", "select": "id,email,display_name,rate_proportion,consolidate_as", }) lookup = { r["email"].lower(): r for r in (tm_rows or []) } print(f" team_members loaded: {len(lookup)}") # Read CSV. Skip BOM if present. with csv_path.open(encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) rows = list(reader) print(f"Parsing {len(rows)} rows...") entries = [] for r in rows: member = (r.get("Member") or r.get("User") or "").strip() email = (r.get("Email") or "").strip().lower() operator = (r.get("Client") or "").strip() or None raw_proj = (r.get("Project") or "").strip() client_part, project_part = split_client_project(raw_proj) desc = (r.get("Description") or "").strip() or None start_d = (r.get("Start date") or "").strip() start_t = (r.get("Start time") or "").strip() end_d = (r.get("Stop date") or r.get("End date") or "").strip() end_t = (r.get("Stop time") or r.get("End time") or "").strip() duration = (r.get("Duration") or "").strip() # Toggl date format: YYYY-MM-DD (ISO). Time: HH:MM:SS (24h). if not (start_d and start_t): continue start_at = f"{start_d} {start_t}" end_at = f"{end_d} {end_t}" if end_d and end_t else None duration_seconds = parse_dur_to_seconds(duration) # Conversion (apply team-member proportion) tm = lookup.get(email) team_member_id = tm["id"] if tm else None if tm: converted_user = tm.get("consolidate_as") or tm["display_name"] converted_seconds = ( round(duration_seconds * float(tm["rate_proportion"])) if duration_seconds is not None else None ) else: converted_user = None converted_seconds = None billout_cost = ( round((converted_seconds / 3600.0) * base_rate, 4) if converted_seconds is not None else None ) entries.append({ "org_id": org_id, "source_user_name": member or None, "source_user_email": email or None, "operator": operator, "client": client_part, "project": project_part, "description": desc, "billable": True, # Toggl Detailed Report omits this — default true "start_at": start_at, "end_at": end_at, "duration_seconds": duration_seconds, "source_rate_usd": None, "source_amount_usd": None, "team_member_id": team_member_id, "converted_user": converted_user, "converted_duration_seconds": converted_seconds, "billout_cost_usd": billout_cost, }) if not entries: print("No entries to import.") return print(f"Parsed {len(entries)} entries.") # Create the batch batch = sb("POST", "/clockify_imports", body={ "org_id": org_id, "imported_by": importer_id, "filename": csv_path.name, "name": BATCH_NAME, "row_count": len(entries), "source": "toggl", "notes": "Initial Toggl import of Rian's 2026 logs", }, prefer="return=representation") batch_id = batch[0]["id"] print(f"Created batch: {batch_id}") # Attach import_id to every entry for e in entries: e["import_id"] = batch_id # Insert (chunked) CHUNK = 500 for i in range(0, len(entries), CHUNK): chunk = entries[i:i + CHUNK] sb("POST", "/time_entries", body=chunk, prefer="return=minimal") print(f" inserted {i + len(chunk)} / {len(entries)}") # Stats summary matched = sum(1 for e in entries if e["team_member_id"]) unmatched = len(entries) - matched unknown_emails = sorted({ e["source_user_email"] for e in entries if not e["team_member_id"] and e["source_user_email"] }) print() print("Done.") print(f" matched (team member found): {matched}") print(f" unmatched (will be blocked): {unmatched}") if unknown_emails: print(f" unknown source emails: {unknown_emails}") print(f" batch_id: {batch_id}") if __name__ == "__main__": main()