115 lines
3.4 KiB
Python
Executable File
115 lines
3.4 KiB
Python
Executable File
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()
|