161 lines
5.2 KiB
Python
161 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Fetch time entries for a ConnectWise customer and calculate total billed hours.
|
|
|
|
Usage:
|
|
python3 fetch-customer-time-entries.py "<Client Name>"
|
|
python3 fetch-customer-time-entries.py --company-id 20172 --days 365
|
|
|
|
Requires ~/.secrets/connectwise (multi-line .env file):
|
|
CW_COMPANY_ID=VNI
|
|
CW_PUBLIC_KEY=+...
|
|
CW_PRIVATE_KEY=...
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
|
|
import requests
|
|
|
|
|
|
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 load_credentials():
|
|
path = os.path.expanduser("~/.secrets/connectwise")
|
|
if not os.path.exists(path):
|
|
print(f"Credential file not found: {path}")
|
|
sys.exit(1)
|
|
|
|
with open(path, "r") as f:
|
|
raw = f.read().strip()
|
|
|
|
keys = {}
|
|
for line in raw.splitlines():
|
|
if "=" in line and not line.startswith("#"):
|
|
k, v = line.split("=", 1)
|
|
keys[k.strip()] = v.strip()
|
|
|
|
company_id = keys.get("CW_COMPANY_ID", "")
|
|
public_key = keys.get("CW_PUBLIC_KEY", "")
|
|
private_key = keys.get("CW_PRIVATE_KEY", "")
|
|
|
|
if not all([company_id, public_key, private_key]):
|
|
print("Missing required keys in credential file")
|
|
sys.exit(1)
|
|
|
|
# Avoid double-plus: public key already starts with '+'
|
|
auth_string = f"{company_id}+{public_key}:{private_key}"
|
|
if "++" in auth_string:
|
|
auth_string = auth_string.replace("++", "+")
|
|
|
|
encoded = base64.b64encode(auth_string.encode()).decode()
|
|
|
|
return {
|
|
"company_id": company_id,
|
|
"auth_header": f"Basic {encoded}",
|
|
"client_id": _load_cw_client_id(),
|
|
}
|
|
|
|
|
|
def build_headers(creds):
|
|
return {
|
|
"Authorization": creds["auth_header"],
|
|
"clientId": creds["client_id"],
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def find_company_by_name(base_url, headers, name):
|
|
"""Search for a company by name using a LIKE query."""
|
|
url = f"{base_url}/company/companies?conditions=name like '%{name}%'"
|
|
resp = requests.get(url, headers=headers, timeout=15)
|
|
resp.raise_for_status()
|
|
companies = resp.json()
|
|
if not companies:
|
|
return None
|
|
if len(companies) > 1:
|
|
print(f"WARNING: Found {len(companies)} companies matching '{name}':")
|
|
for c in companies[:5]:
|
|
print(f" - {c.get('name')} (ID: {c.get('id')})")
|
|
return companies[0]
|
|
|
|
|
|
def fetch_time_entries(base_url, headers, company_id, days_back=365):
|
|
"""Fetch time entries for a company in the last N days."""
|
|
cutoff = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
# CRITICAL: Use square brackets for datetime values in conditions
|
|
conditions = f"company/id={company_id} and dateEntered>[{cutoff}]"
|
|
url = f"{base_url}/time/entries?conditions={conditions}&pageSize=1000"
|
|
|
|
resp = requests.get(url, headers=headers, timeout=30)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Fetch CW time entries for a customer")
|
|
parser.add_argument("name", nargs="?", help="Company name to search for")
|
|
parser.add_argument("--company-id", type=int, help="Exact ConnectWise company ID")
|
|
parser.add_argument("--days", type=int, default=365, help="Number of days back to query")
|
|
args = parser.parse_args()
|
|
|
|
if not args.name and not args.company_id:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
creds = load_credentials()
|
|
headers = build_headers(creds)
|
|
base_url = "https://cw.vistanetinc.com/v4_6_release/apis/3.0"
|
|
|
|
if args.company_id:
|
|
company_id = args.company_id
|
|
company_name = f"ID {company_id}"
|
|
else:
|
|
company = find_company_by_name(base_url, headers, args.name)
|
|
if not company:
|
|
print(f"No company found matching '{args.name}'")
|
|
sys.exit(1)
|
|
company_id = company["id"]
|
|
company_name = company["name"]
|
|
|
|
print(f"Fetching time entries for {company_name} (ID: {company_id}) in last {args.days} days...")
|
|
entries = fetch_time_entries(base_url, headers, company_id, args.days)
|
|
|
|
# CRITICAL: The hour field is 'actualHours', not 'hours'
|
|
total_hours = sum(e.get("actualHours", 0) or 0 for e in entries)
|
|
|
|
print(f"\nTotal entries: {len(entries)}")
|
|
print(f"Total billed hours: {total_hours:.2f}")
|
|
|
|
if entries:
|
|
print("\nRecent entries:")
|
|
for e in entries[:5]:
|
|
ticket = e.get("ticket", {})
|
|
date = e.get("dateEntered", "")[:10]
|
|
hours = e.get("actualHours", 0) or 0
|
|
notes = (e.get("notes") or "No notes")[:60]
|
|
print(f" Ticket #{ticket.get('id', 'N/A')}: {hours}h on {date} — {notes}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|