99 lines
3.9 KiB
Python
99 lines
3.9 KiB
Python
import json
|
|
import os
|
|
from datetime import datetime
|
|
|
|
har_path = '/home/naeel/terra/har/faststart.har'
|
|
|
|
def analyze_har():
|
|
with open(har_path, 'r', encoding='utf-8') as f:
|
|
har_data = json.load(f)
|
|
|
|
entries = har_data['log']['entries']
|
|
|
|
instance_id = None
|
|
operation_id = None
|
|
start_time = None
|
|
end_time = None
|
|
error_message = None
|
|
|
|
# Sort entries by start time
|
|
entries.sort(key=lambda x: x['startedDateTime'])
|
|
|
|
for entry in entries:
|
|
req = entry['request']
|
|
res = entry['response']
|
|
url = req['url']
|
|
|
|
# Look for instance UID in body
|
|
if req.get('postData', {}).get('text'):
|
|
if 'instanceUid' in req['postData']['text']:
|
|
try:
|
|
body = json.loads(req['postData']['text'])
|
|
instance_id = body.get('instanceUid')
|
|
except: pass
|
|
|
|
# Look for operation creation or status
|
|
if '/instanceOperations' in url:
|
|
if req['method'] == 'POST' and res['status'] == 201:
|
|
try:
|
|
body = json.loads(res['content'].get('text', '{}'))
|
|
operation_id = body.get('uid')
|
|
start_time = entry['startedDateTime']
|
|
print(f"Operation started: {operation_id} at {start_time}")
|
|
except: pass
|
|
|
|
# Match the UUID from the URL if not found in POST
|
|
if not operation_id:
|
|
parts = url.split('/')
|
|
for part in parts:
|
|
if len(part) == 36 and '-' in part: # UUID shape
|
|
operation_id = part
|
|
break
|
|
|
|
# Track operation progress/status
|
|
if operation_id and f'/instanceOperations/{operation_id}' in url and req['method'] == 'GET':
|
|
try:
|
|
text = res['content'].get('text', '{}')
|
|
if not text: continue
|
|
body = json.loads(text)
|
|
if isinstance(body, list):
|
|
body = body[0] if body else {}
|
|
status = body.get('status')
|
|
print(f"Polling {operation_id}: status={status} at {entry['startedDateTime']}")
|
|
# print(f"DEBUG: {text[:200]}") # Print first 200 chars
|
|
if status in ['failed', 'completed', 'success', 'error', 'failed-manual-fix']:
|
|
if not end_time or entry['startedDateTime'] > end_time:
|
|
end_time = entry['startedDateTime']
|
|
error_message = body.get('error') or body.get('message') or body.get('statusMessage')
|
|
if not error_message and body.get('steps'):
|
|
# Check steps for errors
|
|
for step in body.get('steps'):
|
|
if step.get('status') == 'failed':
|
|
error_message = step.get('error') or step.get('statusMessage')
|
|
print(f"Operation reached terminal status {status} at {end_time}")
|
|
print(f"Full response: {text}")
|
|
except Exception as e:
|
|
# print(f"Error parsing: {e}")
|
|
pass
|
|
|
|
# Check for Sun/Date anomalies in headers
|
|
for header in res['headers']:
|
|
if header['name'].lower() == 'date':
|
|
if 'Sun' in header['value']:
|
|
print(f"Strange Date header found: {header['value']} in {url}")
|
|
|
|
print(f"\nSummary:")
|
|
print(f"Instance ID: {instance_id}")
|
|
print(f"Operation ID: {operation_id}")
|
|
if start_time and end_time:
|
|
t1 = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
|
|
t2 = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
|
|
duration = (t2 - t1).total_seconds()
|
|
print(f"Duration: {duration} seconds")
|
|
else:
|
|
print("Could not determine duration")
|
|
print(f"Error: {error_message}")
|
|
|
|
if __name__ == "__main__":
|
|
analyze_har()
|