94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
import requests
|
|
import re
|
|
import json
|
|
import os
|
|
import base64
|
|
from requests.auth import HTTPBasicAuth
|
|
|
|
# Configuration
|
|
BASE_URL = "https://cw.vistanetinc.com/v4_6_release/apis/3.0"
|
|
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"
|
|
)
|
|
|
|
CLIENT_ID = _load_cw_client_id()
|
|
SECRETS_PATH = os.path.expanduser("~/.secrets/connectwise")
|
|
|
|
def get_auth():
|
|
with open(SECRETS_PATH, 'r') as f:
|
|
lines = f.readlines()
|
|
creds = {}
|
|
for line in lines:
|
|
if '=' in line:
|
|
k, v = line.strip().split('=', 1)
|
|
creds[k] = v
|
|
|
|
auth_str = f"{creds['CW_COMPANY_ID']}{creds['CW_PUBLIC_KEY']}:{creds['CW_PRIVATE_KEY']}"
|
|
return auth_str
|
|
|
|
def clean_json(text):
|
|
if not text:
|
|
return text
|
|
return re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text)
|
|
|
|
def cw_request(method, endpoint, params=None, json_data=None):
|
|
url = f"{BASE_URL}/{endpoint.lstrip('/')}"
|
|
auth_str = get_auth()
|
|
|
|
headers = {
|
|
"clientId": CLIENT_ID,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
|
|
encoded = base64.b64encode(auth_str.encode()).decode()
|
|
headers["Authorization"] = f"Basic {encoded}"
|
|
|
|
response = requests.request(
|
|
method=method,
|
|
url=url,
|
|
params=params,
|
|
json=json_data,
|
|
headers=headers,
|
|
timeout=30
|
|
)
|
|
|
|
cleaned_text = clean_json(response.text)
|
|
try:
|
|
return json.loads(cleaned_text), response.status_code
|
|
except json.JSONDecodeError:
|
|
return cleaned_text, response.status_code
|
|
|
|
def get_ticket(ticket_id):
|
|
return cw_request("GET", f"service/tickets/{ticket_id}")
|
|
|
|
def list_tickets(conditions=None, page_size=100):
|
|
params = {"pageSize": page_size}
|
|
if conditions:
|
|
params["conditions"] = conditions
|
|
return cw_request("GET", "service/tickets", params=params)
|
|
|
|
def update_ticket(ticket_id, patch_data):
|
|
return cw_request("PATCH", f"service/tickets/{ticket_id}", json_data=patch_data)
|
|
|
|
def create_schedule_entry(data):
|
|
return cw_request("POST", "schedule/entries", json_data=data)
|
|
|
|
if __name__ == "__main__":
|
|
print("Testing CW Connection...")
|
|
data, status = get_ticket(121216)
|
|
print(f"Status: {status}")
|
|
print(f"Data: {data}")
|