Import time off data from a CSV file via API

pip3 install pandas requests python-dateutil

person_name, timeoff_type_name, start_date, end_date, hours, full_day, notes

#!/usr/bin/env python3
import hashlib
import json
import time
from datetime import datetime

import pandas as pd
import requests

# === Step 1: CSV + Auth Config ===
CSV_FILE = "import_file_name.csv"  # your time off CSV file
DRY_RUN = False
SHOW_SUCCESS_OUTPUT = False
BEARER_TOKEN = "Float API Key"

# === Error log file ===
ERROR_LOG_FILE = "upload-errors-timeoff.txt"

# Initialize error log with start timestamps (local + UTC)
with open(ERROR_LOG_FILE, "w") as f:
    f.write("=== Float Time Off Import Errors ===\n")
    f.write(
        "Import started: "
        f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (local), "
        f"{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC\n\n"
    )

# === Step 2: Float API URLs ===
BASE_URL = "https://api.float.com/v3"
PEOPLE_API_URL = f"{BASE_URL}/people"
TIMEOFF_TYPES_API_URL = f"{BASE_URL}/timeoff-types"
TIMEOFF_API_URL = f"{BASE_URL}/timeoffs"

PER_PAGE = 200

HEADERS = {
    "Authorization": f"Bearer {BEARER_TOKEN}",
    "Content-Type": "application/json",
    "User-Agent": "Float Time Off Importer (support@float.com)",
}

# === Utility Functions ===
def sanitize(value):
    """Return a trimmed string or empty string if value is NaN/blank."""
    if pd.isna(value) or str(value).strip() in ("", "NaN", "nan"):
        return ""
    return str(value).strip()


def log_error(row_num, message, payload=None):
    """Append an error entry to upload-errors file."""
    with open(ERROR_LOG_FILE, "a") as f:
        f.write(f"Row {row_num}: {message}\n")
        if payload:
            try:
                f.write(f"Payload: {json.dumps(payload, indent=2)}\n")
            except Exception:
                f.write(f"Payload: {str(payload)}\n")
        f.write("\n")


def generate_color(key: str) -> str:
    """Deterministically generate Float-compatible color (RRGGBB / no '#')."""
    if not key:
        return "CCCCCC"
    h = hashlib.sha1(key.encode("utf-8")).hexdigest()
    return h[:6].upper()


def fetch_all(endpoint, label):
    """Fetch paginated results from Float."""
    try:
        print(f"📦 Fetching existing {label} ...")
        cur_page, last_page = 1, 1
        items = []

        while cur_page <= last_page:
            resp = requests.get(
                f"{endpoint}?per-page={PER_PAGE}&page={cur_page}",
                headers=HEADERS,
            )
            resp.raise_for_status()

            last_page = int(resp.headers.get("x-pagination-page-count") or 1)
            items.extend(resp.json())
            cur_page += 1

        print(f"→ Found {len(items)} {label}")
        return items

    except Exception as e:
        msg = f"Failed to fetch {label}: {e}"
        print("❌ " + msg)
        log_error("FETCH", msg)
        return []


# === Step 5: Load CSV File ===
try:
    df = pd.read_csv(CSV_FILE)
except Exception as e:
    print(f"❌ Failed to read CSV: {e}")
    log_error("CSV", f"Failed to read CSV: {e}")
    raise SystemExit(1)

required_cols = ["person_name", "timeoff_type_name", "start_date", "end_date", "hours"]
missing = [c for c in required_cols if c not in df.columns]

if missing:
    msg = f"CSV missing required columns: {', '.join(missing)}"
    print(f"❌ {msg}")
    log_error("CSV", msg)
    raise SystemExit(1)

print("✅ CSV validation passed.\n")

# === Step 6: Fetch Existing Data ===
people = fetch_all(PEOPLE_API_URL, "people")
timeoff_types = fetch_all(TIMEOFF_TYPES_API_URL, "time off types")

# === Step 7: Build Lookup Maps ===
people_map = {sanitize(p["name"]).lower(): p for p in people}
timeoff_types_map = {
    sanitize(t["timeoff_type_name"]).lower(): t for t in timeoff_types
}

print("\n🚀 Starting Time Off import...\n")

# === Step 8: Main Import Loop ===
for index, row in df.iterrows():
    row_num = index + 1

    try:
        # === PERSON ===
        person_name = sanitize(row.get("person_name"))
        if not person_name:
            raise Exception("Missing person_name")

        pkey = person_name.lower()
        if pkey not in people_map:
            raise Exception(f"Unknown person '{person_name}'")

        person_id = people_map[pkey]["people_id"]

        # === TIME OFF TYPE ===
        tot_name = sanitize(row.get("timeoff_type_name"))
        if not tot_name:
            raise Exception("Missing timeoff_type_name")

        tkey = tot_name.lower()

        if tkey in timeoff_types_map:
            timeoff_type_id = timeoff_types_map[tkey]["timeoff_type_id"]
        else:
            new_type_payload = {
                "timeoff_type_name": tot_name,
                "color": generate_color(tot_name),
                "active": 1,
            }

            if DRY_RUN:
                print(f"[Dry-run] Creating timeoff type: {new_type_payload}")
                timeoff_type_id = 900000 + row_num
            else:
                resp = requests.post(
                    TIMEOFF_TYPES_API_URL,
                    json=new_type_payload,
                    headers=HEADERS,
                )
                if resp.ok:
                    t = resp.json()
                    timeoff_type_id = t["timeoff_type_id"]
                    timeoff_types_map[tkey] = t
                else:
                    raise Exception(
                        f"POST timeoff-types failed: "
                        f"{resp.status_code} - {resp.text}"
                    )

        # === FULL DAY + HOURS LOGIC ===
        full_day_raw = row.get("full_day")
        full_day = int(full_day_raw) if not pd.isna(full_day_raw) else None
        is_full_day = full_day == 1

        hours_raw = row.get("hours")
        hours_num = float(hours_raw) if not pd.isna(hours_raw) else None

        # If full-day time off: allow missing hours or hours below 0.01
        if is_full_day:
            if hours_num is not None and hours_num < 0.01:
                hours_num = None

        # === TIMEOFF PAYLOAD ===
        repeat_state = row.get("repeat_state")
        repeat_end = sanitize(row.get("repeat_end_date"))
        status_val = row.get("status")
        notes = sanitize(row.get("timeoff_notes"))

        timeoff_payload = {
            "timeoff_type_id": timeoff_type_id,
            "people_ids": [person_id],
            "start_date": sanitize(row.get("start_date")),
            "end_date": sanitize(row.get("end_date")),
            "timeoff_notes": notes or None,
            "repeat_state": int(repeat_state)
            if not pd.isna(repeat_state)
            else None,
            "repeat_end": repeat_end or None,
            "status": int(status_val) if not pd.isna(status_val) else None,
            "full_day": full_day if full_day is not None else None,
        }

        # Add hours only when appropriate
        if hours_num is not None:
            timeoff_payload["hours"] = hours_num

        # Remove None values
        timeoff_payload = {k: v for k, v in timeoff_payload.items() if v is not None}

        if DRY_RUN:
            print(f"[Dry-run] Creating timeoff: {timeoff_payload}")
        else:
            resp = requests.post(
                TIMEOFF_API_URL,
                json=timeoff_payload,
                headers=HEADERS,
            )
            if resp.ok:
                if SHOW_SUCCESS_OUTPUT:
                    print(f"✅ Time off created for Row {row_num}")
            else:
                raise Exception(
                    f"POST timeoffs failed: {resp.status_code} - {resp.text}"
                )

        time.sleep(0.75)

    except Exception as e:
        msg = str(e)
        print(f"❌ Error in Row {row_num}: {msg}")
        log_error(row_num, msg)

# === Completion ===
print("\n🎉 Time Off Import completed.")
print(f"📝 Errors (if any) logged to: {ERROR_LOG_FILE}")

with open(ERROR_LOG_FILE, "a") as f:
    f.write(
        "\nImport completed: "
        f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (local), "
        f"{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC\n"
    )