Initial commit

2026-04-28 09:44:22 +10:00
commit 87bfe26890
11 changed files with 1252 additions and 0 deletions
@@ -0,0 +1,114 @@
+import os
+import sys
+from typing import Any, Dict, List
+
+import requests
+
+BASE_URL = os.getenv('SERVICEM8_BASE_URL', 'https://api.servicem8.com')
+ACCESS_TOKEN = os.getenv('SERVICEM8_ACCESS_TOKEN', 'smk-ac525b-99c4b96305a49c7c-fe4dd3e705b647ea')
+STATUS = os.getenv('SERVICEM8_WEBHOOK_STATUS', 'all')
+ENDPOINT = '/webhook_subscriptions'
+
+
+def normalize(item: Dict[str, Any]) -> Dict[str, Any]:
+    sub_type = item.get('type', '-')
+    target = item.get('event') if sub_type == 'event' else item.get('object', '-')
+    fields = ','.join(item.get('fields', [])) if sub_type == 'object' else '-'
+    active = bool(item.get('active', False))
+    failure_reason = item.get('last_failure_reason') or ''
+    failure_at = item.get('last_failure_at') or ''
+    callback_url = item.get('callback_url', '-')
+    unique_id = item.get('unique_id') or '-'
+    status_label = 'ACTIVE' if active else 'INACTIVE'
+    issue = 'FAIL' if failure_reason else ''
+    sort_bucket = 0 if (not active or failure_reason) else 1
+
+    return {
+        'status': status_label,
+        'issue': issue,
+        'type': sub_type,
+        'target': target or '-',
+        'fields': fields or '-',
+        'unique_id': unique_id,
+        'callback_url': callback_url,
+        'failure_at': failure_at or '-',
+        'failure_reason': failure_reason or '-',
+        '_sort_bucket': sort_bucket,
+    }
+
+
+def truncate(value: str, width: int) -> str:
+    value = str(value)
+    return value if len(value) <= width else value[: width - 3] + '...'
+
+
+def render_table(rows: List[Dict[str, Any]]) -> None:
+    columns = [
+        ('status', 'STATUS', 9),
+        ('issue', 'ISSUE', 5),
+        ('type', 'TYPE', 8),
+        ('target', 'TARGET', 24),
+        ('fields', 'FIELDS', 24),
+        ('unique_id', 'UNIQUE_ID', 18),
+        ('failure_at', 'FAILURE_AT', 19),
+        ('callback_url', 'CALLBACK_URL', 42),
+    ]
+
+    header = ' | '.join(title.ljust(width) for _, title, width in columns)
+    divider = '-+-'.join('-' * width for _, _, width in columns)
+    print(header)
+    print(divider)
+
+    for row in rows:
+        line = ' | '.join(truncate(row[key], width).ljust(width) for key, _, width in columns)
+        print(line)
+
+
+def main() -> None:
+    if STATUS not in {'active', 'inactive', 'all'}:
+        print('SERVICEM8_WEBHOOK_STATUS must be one of: active, inactive, all')
+        sys.exit(1)
+
+    url = f'{BASE_URL}{ENDPOINT}'
+    headers = {
+        'X-API-Key': f'{ACCESS_TOKEN}',
+        'Accept': 'application/json',
+    }
+    params = {'status': STATUS}
+
+    response = requests.get(url, headers=headers, params=params, timeout=30)
+    print(f'HTTP {response.status_code}')
+
+    try:
+        data = response.json()
+    except ValueError:
+        print('Response was not valid JSON:')
+        print(response.text)
+        sys.exit(1)
+
+    if not response.ok:
+        print(data)
+        sys.exit(1)
+
+    rows = [normalize(item) for item in data]
+    rows.sort(key=lambda r: (r['_sort_bucket'], r['status'], r['type'], r['target']))
+
+    print(f'Subscriptions returned: {len(rows)}')
+    print(f'Filter: {STATUS}')
+    print()
+
+    if not rows:
+        print('No webhook subscriptions found.')
+        return
+
+    render_table(rows)
+
+    failures = [r for r in rows if r['failure_reason'] != '-']
+    if failures:
+        print('\nFailure details:')
+        for row in failures:
+            print(f"- {row['type']} {row['target']} -> {row['failure_reason']} (at {row['failure_at']})")
+
+
+if __name__ == '__main__':
+    main()