94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
import os
|
|
import requests, base64, json, datetime, re
|
|
|
|
# Generic ConnectWise cert-renewal scheduler.
|
|
# Usage: edit TICKETS list, run. Schedule = expiry minus 7 days;
|
|
# Sat/Sun shifts to preceding Friday; 9am PDT start, staggered +1h per same-day block.
|
|
# Set PATCH_STATUS=True to flip tickets to 'Scheduled' (status id 13) after entry creation.
|
|
|
|
def _load_creds():
|
|
"""Load ConnectWise API creds from ~/.secrets/connectwise or env."""
|
|
path = os.path.expanduser("~/.secrets/connectwise")
|
|
keys = {}
|
|
if os.path.exists(path):
|
|
with open(path) as f:
|
|
for line in f:
|
|
if "=" in line and not line.lstrip().startswith("#"):
|
|
k, v = line.strip().split("=", 1)
|
|
keys[k.strip()] = v.strip()
|
|
for k in ("CW_COMPANY_ID", "CW_PUBLIC_KEY", "CW_PRIVATE_KEY", "CW_CLIENT_ID"):
|
|
if os.getenv(k):
|
|
keys[k] = os.environ[k]
|
|
missing = [
|
|
k for k in ("CW_COMPANY_ID", "CW_PUBLIC_KEY", "CW_PRIVATE_KEY", "CW_CLIENT_ID")
|
|
if not keys.get(k)
|
|
]
|
|
if missing:
|
|
raise SystemExit(
|
|
f"missing {missing}: export them or populate ~/.secrets/connectwise"
|
|
)
|
|
return keys
|
|
|
|
|
|
_CREDS = _load_creds()
|
|
COMPANY_ID = _CREDS["CW_COMPANY_ID"]
|
|
PUBLIC_KEY = _CREDS["CW_PUBLIC_KEY"].lstrip("+")
|
|
PRIVATE_KEY = _CREDS["CW_PRIVATE_KEY"]
|
|
CLIENT_ID = _CREDS["CW_CLIENT_ID"]
|
|
|
|
BASE = "https://cw.vistanetinc.com/v4_6_release/apis/3.0"
|
|
MEMBER_ID = 147 # SBrown
|
|
WHERE_ID = 4 # Remote
|
|
SCHED_STATUS_ID = 2 # Firm
|
|
SCHED_TYPE_ID = 4 # 'S'
|
|
SPAN_ID = 1 # normal block
|
|
REMINDER_ID = 4 # 15 minutes
|
|
TICKET_STATUS_SCHEDULED = 13
|
|
PATCH_STATUS = False # True -> also PATCH ticket status to Scheduled
|
|
|
|
auth = base64.b64encode(f"{COMPANY_ID}+{PUBLIC_KEY}:{PRIVATE_KEY}".encode()).decode()
|
|
H = {"Authorization": f"Basic {auth}", "clientId": CLIENT_ID, "Content-Type": "application/json"}
|
|
|
|
TICKETS = [] # e.g. [122333, 122334, 122332, 122335, 122331, 122330, 122329, 122328]
|
|
|
|
|
|
def plan_date(expiry):
|
|
d = datetime.date.fromisoformat(expiry) - datetime.timedelta(days=7)
|
|
while d.weekday() >= 5:
|
|
d -= datetime.timedelta(days=1)
|
|
return d
|
|
|
|
|
|
def main():
|
|
plan, slot = [], {}
|
|
for tid in TICKETS:
|
|
t = requests.get(f"{BASE}/service/tickets/{tid}", headers=H).json()
|
|
m = re.search(r"Expiring on (\d{4}-\d{2}-\d{2})", t.get("summary") or "")
|
|
expiry = m.group(1)
|
|
d = plan_date(expiry)
|
|
hour = 16 + slot.get(d, 0)
|
|
slot[d] = slot.get(d, 0) + 1
|
|
plan.append({"tid": tid, "co": t["company"]["identifier"], "expiry": expiry,
|
|
"date": d.isoformat(),
|
|
"start": f"{d.isoformat()}T{hour:02d}:00:00Z",
|
|
"end": f"{d.isoformat()}T{hour+1:02d}:00:00Z"})
|
|
for p in plan:
|
|
r = requests.get(f"{BASE}/schedule/entries", headers=H,
|
|
params={"conditions": f"(objectId = {p['tid']})", "pageSize": 50})
|
|
if any(e.get("dateStart") for e in r.json()):
|
|
print(f"{p['tid']}: SKIP (existing dated block)"); continue
|
|
body = {"name": f"{p['co']} / Domain certificate renewal expiring {p['expiry']}",
|
|
"objectId": p["tid"], "member": {"id": MEMBER_ID}, "where": {"id": WHERE_ID},
|
|
"dateStart": p["start"], "dateEnd": p["end"], "reminder": {"id": REMINDER_ID},
|
|
"status": {"id": SCHED_STATUS_ID}, "type": {"id": SCHED_TYPE_ID},
|
|
"span": {"id": SPAN_ID}, "hours": 1.0}
|
|
r = requests.post(f"{BASE}/schedule/entries", headers=H, json=body)
|
|
print(f"{p['tid']}: {r.status_code} {r.json().get('id') if r.status_code == 201 else r.text[:150]}")
|
|
if PATCH_STATUS:
|
|
requests.patch(f"{BASE}/service/tickets/{p['tid']}", headers=H,
|
|
json=[{"op": "replace", "path": "status/id", "value": TICKET_STATUS_SCHEDULED}])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|