| import argparse |
| import sys |
| import os |
| import json |
| import requests |
|
|
| sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) |
|
|
| from app.core.config import settings |
|
|
| def check_health(base_url: str): |
| print(f"[*] Checking system health at {base_url}...") |
| try: |
| r = requests.get(f"{base_url}/health", timeout=5) |
| if r.status_code == 200: |
| print("[✓] Health Check PASSED:", json.dumps(r.json(), indent=2)) |
| else: |
| print(f"[✗] Health Check FAILED with status code {r.status_code}") |
| except Exception as e: |
| print(f"[✗] Connection Error: {str(e)}") |
|
|
| def check_metrics(base_url: str): |
| print(f"[*] Fetching Prometheus metrics at {base_url}/metrics...") |
| try: |
| r = requests.get(f"{base_url}/metrics", timeout=5) |
| if r.status_code == 200: |
| lines = r.text.splitlines() |
| http_reqs = [l for l in lines if "http_requests_total" in l and not l.startswith("#")] |
| print("[✓] Prometheus Metrics Endpoint Online.") |
| print(f" Active Http Request Metrics Snippet ({len(http_reqs)} entries found):") |
| for line in http_reqs[:5]: |
| print(f" - {line}") |
| else: |
| print(f"[✗] Metrics Endpoint returned status code {r.status_code}") |
| except Exception as e: |
| print(f"[✗] Metrics Error: {str(e)}") |
|
|
| def main(): |
| parser = argparse.ArgumentParser(description="12D AI Assessment Engine Ops CLI Utility") |
| parser.add_argument("command", choices=["health", "metrics", "version"], help="Command to execute") |
| parser.add_argument("--url", default="http://localhost:8000", help="Base URL of the Assessment Engine API") |
|
|
| args = parser.parse_args() |
|
|
| if args.command == "health": |
| check_health(args.url) |
| elif args.command == "metrics": |
| check_metrics(args.url) |
| elif args.command == "version": |
| print("12D AI Assessment Engine Version: 1.0.0-rc1") |
|
|
| if __name__ == "__main__": |
| main() |
|
|
