148 lines
4.6 KiB
Python
148 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cross-company top N by billed hours — past year or custom range.
|
|
|
|
Usage:
|
|
python top-companies-by-hours.py [DAYS] [TOP_N]
|
|
|
|
Arguments:
|
|
DAYS Number of days back to query (default: 365)
|
|
TOP_N Number of companies to show (default: 30)
|
|
|
|
Outputs a ranked table with company name, lead tech (whoever logged
|
|
most hours), and total billed actualHours.
|
|
|
|
Auth: parses ~/.secrets/connectwise (CW_COMPANY_ID, CW_PUBLIC_KEY, CW_PRIVATE_KEY)
|
|
"""
|
|
|
|
import base64
|
|
import requests
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
from collections import defaultdict
|
|
import os
|
|
|
|
|
|
def load_creds(path="~/.secrets/connectwise"):
|
|
with open(os.path.expanduser(path), "r") as f:
|
|
lines = f.read().strip().split("\n")
|
|
creds = {}
|
|
for line in lines:
|
|
if "=" in line:
|
|
k, v = line.split("=", 1)
|
|
creds[k] = v
|
|
return creds
|
|
|
|
|
|
def _load_cw_client_id():
|
|
"""Resolve the ConnectWise clientId from env or ~/.secrets/connectwise."""
|
|
cid = os.getenv("CW_CLIENT_ID", "")
|
|
if cid:
|
|
return cid
|
|
path = os.path.expanduser("~/.secrets/connectwise")
|
|
if os.path.exists(path):
|
|
with open(path) as f:
|
|
for line in f:
|
|
if line.lstrip().startswith("CW_CLIENT_ID="):
|
|
return line.split("=", 1)[1].strip()
|
|
raise SystemExit(
|
|
"CW_CLIENT_ID missing: export CW_CLIENT_ID or add "
|
|
"CW_CLIENT_ID=<uuid> to ~/.secrets/connectwise"
|
|
)
|
|
|
|
|
|
def get_headers(creds):
|
|
company_id = creds["CW_COMPANY_ID"]
|
|
public_key = creds["CW_PUBLIC_KEY"]
|
|
private_key = creds["CW_PRIVATE_KEY"]
|
|
auth_str = f"{company_id}{public_key}:{private_key}"
|
|
return {
|
|
"Authorization": f"Basic {base64.b64encode(auth_str.encode()).decode()}",
|
|
"clientId": _load_cw_client_id(),
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def fetch_time_entries(headers, days_back=365):
|
|
base_url = "https://cw.vistanetinc.com/v4_6_release/apis/3.0"
|
|
cutoff = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
url = f"{base_url}/time/entries?conditions=timeStart%3E[{cutoff}]"
|
|
|
|
all_entries = []
|
|
page = 1
|
|
page_size = 1000
|
|
max_pages = 20
|
|
|
|
while page <= max_pages:
|
|
paged_url = f"{url}&pageSize={page_size}&page={page}"
|
|
try:
|
|
resp = requests.get(paged_url, headers=headers, timeout=30)
|
|
if resp.status_code != 200:
|
|
print(f"API error {resp.status_code}: {resp.text[:200]}", file=sys.stderr)
|
|
break
|
|
data = resp.json()
|
|
entries = data if isinstance(data, list) else data.get("data", [])
|
|
if not entries:
|
|
break
|
|
all_entries.extend(entries)
|
|
if len(entries) < page_size:
|
|
break
|
|
page += 1
|
|
except Exception as e:
|
|
print(f"Request error: {e}", file=sys.stderr)
|
|
break
|
|
|
|
return all_entries, cutoff
|
|
|
|
|
|
def aggregate(entries):
|
|
company_hours = defaultdict(float)
|
|
company_names = {}
|
|
company_member_hours = defaultdict(lambda: defaultdict(float))
|
|
|
|
for entry in entries:
|
|
comp = entry.get("company", {})
|
|
comp_id = comp.get("id")
|
|
if comp_id is None:
|
|
continue
|
|
hours = entry.get("actualHours", 0) or 0
|
|
company_hours[comp_id] += hours
|
|
company_names[comp_id] = comp.get("name", "Unknown")
|
|
|
|
member = entry.get("member", {})
|
|
member_name = member.get("name", "Unknown")
|
|
if member_name and member_name != "Unknown":
|
|
company_member_hours[comp_id][member_name] += hours
|
|
|
|
lead_techs = {
|
|
comp_id: max(members.items(), key=lambda x: x[1])[0]
|
|
for comp_id, members in company_member_hours.items()
|
|
}
|
|
return company_hours, company_names, lead_techs
|
|
|
|
|
|
def main():
|
|
days = int(sys.argv[1]) if len(sys.argv) > 1 else 365
|
|
top_n = int(sys.argv[2]) if len(sys.argv) > 2 else 30
|
|
|
|
creds = load_creds()
|
|
headers = get_headers(creds)
|
|
|
|
print(f"Fetching time entries from {days} days ago...", file=sys.stderr)
|
|
entries, cutoff = fetch_time_entries(headers, days)
|
|
print(f"Retrieved {len(entries)} entries", file=sys.stderr)
|
|
|
|
company_hours, company_names, lead_techs = aggregate(entries)
|
|
ranked = sorted(company_hours.items(), key=lambda x: x[1], reverse=True)[:top_n]
|
|
|
|
print(f"\n{'Rank':<6}{'Company':<50}{'Lead Tech':<25}{'Hours':>10}")
|
|
print("-" * 91)
|
|
for rank, (comp_id, hours) in enumerate(ranked, 1):
|
|
name = company_names.get(comp_id, "Unknown")
|
|
lead = lead_techs.get(comp_id, "N/A")
|
|
print(f"{rank:<6}{name:<50}{lead:<25}{hours:>10.2f}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|