Pull Reporting into a BI Tool

import base64
import csv
import os
import time

import requests

BASE_URL = "https://portal.contraforce.com/api/v2"
WORKSPACES = {
    "3f2b8c1e-7d4a-4e5b-9a61-2c8f0d7e4b19": "Contoso Retail",
    "9a7d2e64-1b3c-4f8e-b5a2-6e0c4d1f8a73": "Fabrikam Health",
}

credential = f"{os.environ['CONTRAFORCE_CLIENT_ID']}:{os.environ['CONTRAFORCE_CLIENT_SECRET']}"
session = requests.Session()
session.headers["Authorization"] = "Basic " + base64.b64encode(credential.encode()).decode()


def report(path, workspace_id):
    body = {"from": "2026-01-01", "to": "2026-10-31", "workspaceIds": [workspace_id]}
    while True:
        response = session.post(f"{BASE_URL}/reporting/{path}", json=body, timeout=60)
        if response.status_code == 429:
            time.sleep(60)  # The budget refills every minute.
            continue
        response.raise_for_status()
        return response.json()["data"]


with open("reporting-summary.csv", "w", newline="") as file:
    writer = csv.writer(file)
    writer.writerow([
        "workspace", "month", "received", "closed", "closed_by_agent", "closed_by_analyst",
        "investigated", "median_hours_to_close", "agent_runs", "resolved_by_agent", "escalated",
    ])
    for workspace_id, name in WORKSPACES.items():
        flow = report("incident-flow", workspace_id)
        runs_by_month = {point["date"]: point for point in report("agent-investigations", workspace_id)["points"]}
        for point in flow["points"]:
            runs = runs_by_month.get(point["date"], {})
            median = point["medianTimeToCloseSeconds"]
            writer.writerow([
                name,
                point["date"],
                point["received"],
                point["closed"],
                point["closedBy"].get("Agent", 0),
                point["closedBy"].get("Analyst", 0),
                point["investigated"],
                round(median / 3600, 1) if median is not None else "",
                runs.get("runs", 0),
                runs.get("resolvedByAgent", 0),
                runs.get("escalated", 0),
            ])