"""Comprehensive Service Inspector and Diagnostic Tool for DSCons.""" import json import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import socket import ssl import urllib.error import urllib.request from datetime import datetime, timezone import psycopg from qdrant_client import QdrantClient from app.core.settings import get_settings def inspect_all(): settings = get_settings() results = { "timestamp": datetime.now(timezone.utc).isoformat(), "status": "HEALTHY", "checks": {}, } # 1. Database Check db_res = {"status": "UNKNOWN", "details": {}} db_url = getattr(settings, "database_url", None) or os.environ.get( "DATABASE_URL", "postgresql://postgres:postgres@127.0.0.1:5432/dscons" ) try: with psycopg.connect(db_url, connect_timeout=3) as conn: with conn.cursor() as cur: cur.execute("SELECT current_database(), current_user, version();") db_info = cur.fetchone() cur.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' ORDER BY table_name; """) tables = [row[0] for row in cur.fetchall()] db_res["status"] = "OK" db_res["details"] = { "database": db_info[0], "user": db_info[1], "version": db_info[2], "tables_count": len(tables), "tables": tables, } except Exception as e: db_res["status"] = "ERROR" db_res["error"] = str(e) results["status"] = "DEGRADED" results["checks"]["postgresql"] = db_res # 2. Qdrant Vector Engine Check qdrant_res = {"status": "UNKNOWN", "details": {}} try: client = QdrantClient( url=settings.qdrant_url, timeout=5.0, check_compatibility=False ) colls = client.get_collections().collections col_details = {} for c in colls: info = client.get_collection(c.name) points_count = getattr(info, "points_count", None) or getattr( info, "vectors_count", 0 ) col_details[c.name] = { "points_count": points_count, "status": str(getattr(info, "status", "UNKNOWN")), } qdrant_res["status"] = "OK" qdrant_res["details"] = {"url": settings.qdrant_url, "collections": col_details} except Exception as e: qdrant_res["status"] = "ERROR" qdrant_res["error"] = str(e) results["status"] = "DEGRADED" results["checks"]["qdrant"] = qdrant_res # 3. Local FastAPI Backend Check backend_res = {"status": "UNKNOWN", "endpoints": {}} local_endpoints = [ ("liveness", f"http://127.0.0.1:{settings.dscons_port}/health"), ("readiness", f"http://127.0.0.1:{settings.dscons_port}/health/readiness"), ("docs", f"http://127.0.0.1:{settings.dscons_port}/docs"), ("root", f"http://127.0.0.1:{settings.dscons_port}/"), ] all_local_ok = True for name, url in local_endpoints: try: req = urllib.request.Request( url, headers={"User-Agent": "DSCons-Inspector/1.0"} ) with urllib.request.urlopen(req, timeout=5) as response: body = response.read().decode("utf-8", errors="ignore") backend_res["endpoints"][name] = { "url": url, "status_code": response.status, "bytes": len(body), "ok": response.status == 200, } except Exception as e: all_local_ok = False backend_res["endpoints"][name] = {"url": url, "error": str(e), "ok": False} backend_res["status"] = "OK" if all_local_ok else "ERROR" if not all_local_ok: results["status"] = "DEGRADED" results["checks"]["local_backend"] = backend_res # 4. Public Web Connectivity & SSL Check (dinhsonconstruction.com & erp.dinhsonconstruction.com) public_res = {"status": "UNKNOWN", "endpoints": {}, "ssl_info": {}} public_urls = [ ("landing", "https://dinhsonconstruction.com/"), ("health", "https://dinhsonconstruction.com/health"), ("readiness", "https://dinhsonconstruction.com/health/readiness"), ("erp_domain", "https://erp.dinhsonconstruction.com/"), ("erp_dashboard", "https://dinhsonconstruction.com/dashboard"), ("projects", "https://dinhsonconstruction.com/dashboard/projects"), ("equipment", "https://dinhsonconstruction.com/dashboard/equipment"), ("invoices", "https://dinhsonconstruction.com/dashboard/invoices"), ("partners", "https://dinhsonconstruction.com/dashboard/partners"), ("war_room", "https://dinhsonconstruction.com/dashboard/war-room"), ("agents", "https://dinhsonconstruction.com/dashboard/agents"), ] all_pub_ok = True for name, url in public_urls: try: req = urllib.request.Request( url, headers={"User-Agent": "DSCons-Inspector/1.0"} ) with urllib.request.urlopen(req, timeout=8) as response: body = response.read().decode("utf-8", errors="ignore") public_res["endpoints"][name] = { "url": url, "status_code": response.status, "bytes": len(body), "ok": response.status == 200, } except Exception as e: all_pub_ok = False public_res["endpoints"][name] = {"url": url, "error": str(e), "ok": False} # SSL Inspection try: ctx = ssl.create_default_context() with ctx.wrap_socket( socket.socket(), server_hostname="dinhsonconstruction.com" ) as s: s.settimeout(5.0) s.connect(("dinhsonconstruction.com", 443)) cert = s.getpeercert() public_res["ssl_info"] = { "subject": dict(x[0] for x in cert.get("subject", [])), "issuer": dict(x[0] for x in cert.get("issuer", [])), "notBefore": cert.get("notBefore"), "notAfter": cert.get("notAfter"), "version": cert.get("version"), } except Exception as e: public_res["ssl_info"]["error"] = str(e) public_res["status"] = "OK" if all_pub_ok else "ERROR" if not all_pub_ok: results["status"] = "DEGRADED" results["checks"]["public_website"] = public_res # Save Results os.makedirs("reports", exist_ok=True) os.makedirs("storage/logs", exist_ok=True) timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S") json_path = f"storage/logs/system_health_audit_{timestamp_str}.json" latest_json_path = "storage/logs/system_health_latest.json" with open(json_path, "w", encoding="utf-8") as f: json.dump(results, f, indent=2, ensure_ascii=False) with open(latest_json_path, "w", encoding="utf-8") as f: json.dump(results, f, indent=2, ensure_ascii=False) return results, json_path if __name__ == "__main__": results, path = inspect_all() print(json.dumps(results, indent=2, ensure_ascii=False)) print(f"\n[OK] Health audit record saved to: {path}")