Import clients, projects, phases, people, and tasks via CSV + API

python3 --version

pip install pandas requests

project_name,project_code,client_name,project_notes,budget_type,budget_total,budget_priority,status,non_billable,project_tags,phase_name,phase_start_date,phase_end_date,phase_budget_total,person_name,default_hourly_rate,people_tags,task_name,start_date,end_date,hours Website Redesign,WD-001,Acme Corp,Initial rollout,1,200,0,2,0,Marketing,Design Phase,2025-08-01,2025-08-15,100,Jane Smith,85,Remote,UI Design,2025-08-01,2025-08-05,6 Website Redesign,WD-001,Acme Corp,Initial rollout,,,,,,Marketing,,,,,,Jane Smith,85,Remote,UX Research,2025-08-06,2025-08-10,4

PROJECT_UNIQUENESS_STRATEGY = "project_code" # Options: "name", "name+client", "project_code"

#!/usr/bin/env python3
import pandas as pd
import requests
import time
import json
import hashlib
from datetime import datetime

# === Step 1: CSV + Auth Config ===
CSV_FILE = "you_file.csv" #replace with name of your csv file
DRY_RUN = False
SHOW_SUCCESS_OUTPUT = True
BEARER_TOKEN = "Your_API_Token" #Replace with API token from integrations page

# === Error log file ===
ERROR_LOG_FILE = "upload-errors1.txt"
# Initialize error log with start timestamps (local + UTC)
with open(ERROR_LOG_FILE, "w") as f:
    f.write("=== Float Upload 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"
CLIENTS_API_URL = f"{BASE_URL}/clients"
PROJECTS_API_URL = f"{BASE_URL}/projects"
PHASES_API_URL = f"{BASE_URL}/phases"
PEOPLE_API_URL = f"{BASE_URL}/people"
TASKS_API_URL = f"{BASE_URL}/tasks"

# === Step 3: Configuration Options ===
PROJECT_UNIQUENESS_STRATEGY = "name+client"  # Options: "name", "name+client", "project_code"
PER_PAGE = 200
HEADERS = {
    "Authorization": f"Bearer {BEARER_TOKEN}",
    "Content-Type": "application/json",
    "User-Agent": "Float Data Importer (support@float.com)"
}

# === Step 4: Utility Functions ===
def sanitize(value):
    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.txt."""
    with open(ERROR_LOG_FILE, "a") as f:
        f.write(f"Row {row_num}: {message}\n")
        if payload is not None:
            try:
                f.write(f"Payload: {json.dumps(payload, indent=2)}\n")
            except TypeError:
                f.write(f"Payload (unserializable): {str(payload)}\n")
        f.write("\n")

def generate_color(project_key: str) -> str:
    """
    Deterministically generate a Float-compatible hex color (NO # sign).
    Produces a stable 6-digit uppercase hex string like "FF4B4B".
    """
    if not project_key:
        return "CCCCCC"  # Neutral gray fallback
    h = hashlib.sha1(project_key.encode("utf-8")).hexdigest()
    return h[:6].upper()  # Float expects "RRGGBB"
def fetch_existing(endpoint, label):
    """Fetch existing items from Float with pagination."""
    try:
        print(f"📦 Fetching existing {label} from Float...")
        cur_page, last_page = 1, 1
        data = []

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

            last_page = int(response.headers.get("x-pagination-page-count") or 0) or 1
            data.extend(response.json())
            cur_page += 1
        print(f"→ {label.capitalize()}: {len(data)}")
        return data

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

def build_lookup_map(items, key_fn):
    return {key_fn(item): item for item in items if key_fn(item)}

def project_key_from_csv(row):
    name = sanitize(row.get("project_name") or row.get("name"))
    code = sanitize(row.get("project_code"))
    client = sanitize(row.get("client_name"))
    if PROJECT_UNIQUENESS_STRATEGY == "name":
        return name.lower()
    elif PROJECT_UNIQUENESS_STRATEGY == "name+client":
        return f"{client.lower()}|{name.lower()}"
    elif PROJECT_UNIQUENESS_STRATEGY == "project_code":
        return code.lower()
    return None
def project_key_from_api(p, client_lookup):
    name = sanitize(p.get("name"))
    code = sanitize(p.get("project_code"))
    client_id = sanitize(p.get("client_id"))
    client_name = sanitize(client_lookup.get(str(client_id), ""))
    if PROJECT_UNIQUENESS_STRATEGY == "name":
        return name.lower()
    elif PROJECT_UNIQUENESS_STRATEGY == "name+client":
        return f"{client_name.lower()}|{name.lower()}"
    elif PROJECT_UNIQUENESS_STRATEGY == "project_code":
        return code.lower()
    return None
def phase_key(client_id, project_id, phase_name):
    return f"{client_id}|{project_id}|{phase_name.lower()}"
# === 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}")
    # Write completion timestamp before exit
    with open(ERROR_LOG_FILE, "a") as f:
        f.write(
            "\nImport completed (with failure during CSV load): "
            f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (local), "
            f"{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC\n"
        )
    exit(1)

required_columns = ["project_name", "task_name", "person_name", "start_date", "end_date", "hours"]
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
    msg = f"CSV is missing required columns: {', '.join(missing_columns)}"
    print(f"❌ {msg}")
    log_error("CSV", msg)
    with open(ERROR_LOG_FILE, "a") as f:
        f.write(
            "\nImport completed (with failure due to missing columns): "
            f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (local), "
            f"{datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC\n"
        )
    exit(1)
else:
    print("✅ CSV validation passed. All required columns are present.\n")

# === Step 6: Fetch Existing Data ===
clients = fetch_existing("clients", "clients")
projects = fetch_existing("projects", "projects")
people = fetch_existing("people", "people")
phases = fetch_existing("phases", "phases")

# === Step 7: Build Lookup Maps ===
clients_map = build_lookup_map(clients, lambda c: sanitize(c["name"]).lower())
clients_map_rev = {str(v["client_id"]): k for k, v in clients_map.items()}
people_map = build_lookup_map(people, lambda p: sanitize(p["name"]).lower())
projects_map = build_lookup_map(projects, lambda p: project_key_from_api(p, clients_map_rev))
phases_map = build_lookup_map(
    phases,
    lambda ph: phase_key(
        sanitize(ph.get("client_id", "")),
        sanitize(ph.get("project_id", "")),
        sanitize(ph.get("name", ""))
    )
)

print("\n🚀 Starting task import...\n")

# === Step 8: Main Import Loop ===
for index, row in df.iterrows():
    row_num = index + 1
    try:
        # === CLIENT ===
        client_name = sanitize(row.get("client_name"))
        client_id = None
        if client_name:
            client_key = client_name.lower()
            if client_key in clients_map:
                client_id = clients_map[client_key]["client_id"]
            else:
                client_payload = {"name": client_name}
                if DRY_RUN:
                    print(f"[Dry-run] Creating client: {client_payload}")
                    client_id = 999000 + row_num
                else:
                    resp = requests.post(CLIENTS_API_URL, json=client_payload, headers=HEADERS)
                    if resp.ok:
                        client = resp.json()
                        client_id = client["client_id"]
                        clients_map[client_key] = client
                        clients_map_rev[str(client_id)] = client_name
                    else:
                        raise Exception(f"POST clients failed: {resp.status_code} - {resp.text}")

        # === PROJECT ===
        proj_key = project_key_from_csv(row)
        project_id = None
        if proj_key in projects_map:
            project_id = projects_map[proj_key]["project_id"]
        else:
            auto_color = generate_color(proj_key)

            project_payload = {
                "name": sanitize(row.get("project_name")),
                "project_code": sanitize(row.get("project_code")) or None,
                "client_id": client_id if client_id else None,
                "notes": sanitize(row.get("project_notes")),
                "color": auto_color,
                # Optional project-level fields:
                "rate_type": sanitize(row.get("rate_type")) or None,
                "default_hourly_rate": float(row.get("project_default_hourly_rate"))
                if not pd.isna(row.get("project_default_hourly_rate")) else None,
                "budget_type": int(row.get("budget_type")) if not pd.isna(row.get("budget_type")) else None,
                "budget_total": float(row.get("budget_total")) if not pd.isna(row.get("budget_total")) else None,
                "budget_priority": int(row.get("budget_priority")) if not pd.isna(row.get("budget_priority")) else None,
                "status": int(row.get("status")) if not pd.isna(row.get("status")) else None,
                "non_billable": int(row.get("non_billable")) if not pd.isna(row.get("non_billable")) else None,
            }

            if pd.notna(row.get("project_tags")):
                tags = [t.strip() for t in str(row["project_tags"]).split(",") if t.strip()]
                if tags:
                    project_payload["tags"] = tags

            project_payload = {k: v for k, v in project_payload.items() if v is not None}

            if DRY_RUN:
                print(f"[Dry-run] Creating project: {project_payload}")
                project_id = 888000 + row_num
            else:
                resp = requests.post(PROJECTS_API_URL, json=project_payload, headers=HEADERS)
                if resp.ok:
                    project = resp.json()
                    project_id = project["project_id"]
                    projects_map[proj_key] = project
                else:
                    raise Exception(f"POST projects failed: {resp.status_code} - {resp.text}")

        # === PHASE ===
        phase_name = sanitize(row.get("phase_name"))
        phase_id = None
        if phase_name:
            phase_start = sanitize(row.get("phase_start_date"))
            phase_end = sanitize(row.get("phase_end_date"))
            phase_budget = row.get("phase_budget_total")
            phase_status = row.get("phase_status")

            phase_lookup = phase_key(client_id, project_id, phase_name)
            if phase_lookup in phases_map:
                phase_id = phases_map[phase_lookup]["phase_id"]
            else:
                phase_payload = {
                    "name": phase_name,
                    "project_id": project_id,
                    "start_date": phase_start,
                    "end_date": phase_end,
                    "status": int(phase_status) if not pd.isna(phase_status) else None,
                }

                if not pd.isna(phase_budget):
                    phase_payload["budget_total"] = float(phase_budget)

                phase_payload = {k: v for k, v in phase_payload.items() if v is not None}

                if DRY_RUN:
                    print(f"[Dry-run] Creating phase: {phase_payload}")
                    phase_id = 777000 + row_num
                else:
                    resp = requests.post(PHASES_API_URL, json=phase_payload, headers=HEADERS)
                    if resp.ok:
                        phase = resp.json()
                        phase_id = phase["phase_id"]
                        phases_map[phase_lookup] = phase
                    else:
                        raise Exception(f"POST phases failed: {resp.status_code} - {resp.text}")

        # === PERSON ===
        person_name = sanitize(row.get("person_name"))
        person_id = None
        if person_name:
            person_key = person_name.lower()
            if person_key in people_map:
                person_id = people_map[person_key]["people_id"]
            else:
                person_payload = {"name": person_name}

                if not pd.isna(row.get("default_hourly_rate")):
                    person_payload["default_hourly_rate"] = float(row["default_hourly_rate"])

                # Optional person start_date
                if not pd.isna(row.get("person_start_date")):
                    person_payload["start_date"] = sanitize(row.get("person_start_date"))

                if pd.notna(row.get("people_tags")):
                    tags = [t.strip() for t in str(row["people_tags"]).split(",") if t.strip()]
                    if tags:
                        person_payload["tags"] = [{"name": tag} for tag in tags]

                person_payload = {k: v for k, v in person_payload.items() if v is not None}

                if DRY_RUN:
                    print(f"[Dry-run] Creating person: {person_payload}")
                    person_id = 666000 + row_num
                else:
                    resp = requests.post(PEOPLE_API_URL, json=person_payload, headers=HEADERS)
                    if resp.ok:
                        person = resp.json()
                        person_id = person["people_id"]
                        people_map[person_key] = person
                    else:
                        raise Exception(f"POST people failed: {resp.status_code} - {resp.text}")

        # === TASK ===
        task_notes = sanitize(row.get("task_notes"))
        repeat_state = row.get("repeat_state")
        repeat_end_date = sanitize(row.get("repeat_end_date"))

        task_payload = {
            "name": sanitize(row.get("task_name")),
            "project_id": project_id,
            "phase_id": phase_id,
            "people_id": person_id,
            "start_date": sanitize(row.get("start_date")),
            "end_date": sanitize(row.get("end_date")),
            "hours": float(row.get("hours")) if not pd.isna(row.get("hours")) else 0.0,
            "notes": task_notes or None,
            "repeat_state": int(repeat_state) if not pd.isna(repeat_state) else None,
            "repeat_end_date": repeat_end_date or None,
        }

        task_payload = {k: v for k, v in task_payload.items() if v is not None}

        if DRY_RUN:
            print(f"[Dry-run] Creating task: {task_payload}")
        else:
            resp = requests.post(TASKS_API_URL, json=task_payload, headers=HEADERS)
            if resp.ok:
                if SHOW_SUCCESS_OUTPUT:
                    print(f"✅ Task created for Row {row_num}")
            else:
                raise Exception(f"POST tasks failed: {resp.status_code} - {resp.text}")

        time.sleep(0.6)

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

# === Step 9: Wrap up ===
print("\n🎉 Import completed.\n")
print(f"📝 Errors (if any) logged to: {ERROR_LOG_FILE}")

# Append completion timestamps to error log
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"
    )

python3 your_import_script.py