mirror of
https://github.com/acedanger/shell.git
synced 2025-12-06 01:10:12 -08:00
88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import json
|
|
|
|
# Set environment
|
|
os.environ['BACKUP_ROOT'] = '/home/acedanger/shell'
|
|
METRICS_DIR = '/home/acedanger/shell/metrics'
|
|
|
|
|
|
def load_json_file(filepath):
|
|
"""Safely load JSON file with error handling"""
|
|
try:
|
|
if os.path.exists(filepath):
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
|
|
print(f"Error loading JSON file {filepath}: {e}")
|
|
return None
|
|
|
|
|
|
def get_service_metrics(service_name):
|
|
"""Get metrics for a specific service"""
|
|
# Simple status file approach
|
|
status_file = os.path.join(METRICS_DIR, f'{service_name}_status.json')
|
|
|
|
service_status = load_json_file(status_file)
|
|
|
|
return {
|
|
'status': service_status,
|
|
'last_run': service_status.get('end_time') if service_status else None,
|
|
'current_status': service_status.get('status', 'unknown') if service_status else 'never_run',
|
|
'files_processed': service_status.get('files_processed', 0) if service_status else 0,
|
|
'total_size': service_status.get('total_size_bytes', 0) if service_status else 0,
|
|
'duration': service_status.get('duration_seconds', 0) if service_status else 0
|
|
}
|
|
|
|
|
|
def get_consolidated_metrics():
|
|
"""Get consolidated metrics across all services"""
|
|
# With simplified approach, we consolidate by reading all status files
|
|
all_services = {}
|
|
|
|
if os.path.exists(METRICS_DIR):
|
|
for filename in os.listdir(METRICS_DIR):
|
|
if filename.endswith('_status.json'):
|
|
service_name = filename.replace('_status.json', '')
|
|
status_file = os.path.join(METRICS_DIR, filename)
|
|
service_status = load_json_file(status_file)
|
|
if service_status:
|
|
all_services[service_name] = service_status
|
|
|
|
return {
|
|
'services': all_services,
|
|
'total_services': len(all_services),
|
|
'last_updated': '2025-06-18T05:15:00-04:00'
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print('=== Testing Simplified Metrics Web Integration ===')
|
|
|
|
# Test individual service metrics
|
|
print('\n1. Individual Service Metrics:')
|
|
for service in ['plex', 'immich', 'media-services']:
|
|
try:
|
|
metrics = get_service_metrics(service)
|
|
status = metrics['current_status']
|
|
files = metrics['files_processed']
|
|
duration = metrics['duration']
|
|
print(f' {service}: {status} ({files} files, {duration}s)')
|
|
except (OSError, IOError, KeyError) as e:
|
|
print(f' {service}: Error - {e}')
|
|
|
|
# Test consolidated metrics
|
|
print('\n2. Consolidated Metrics:')
|
|
try:
|
|
consolidated = get_consolidated_metrics()
|
|
services = consolidated['services']
|
|
print(f' Total services: {len(services)}')
|
|
for name, status in services.items():
|
|
message = status.get('message', 'N/A')
|
|
print(f' {name}: {status["status"]} - {message}')
|
|
except (OSError, IOError, KeyError) as e:
|
|
print(f' Error: {e}')
|
|
|
|
print('\n✅ Web integration test completed successfully!')
|