vmis/backend/app/services/system_checker.py

"""
SystemChecker — 功能驗證（不只 handshake）
traefik: routers > 0 / keycloak: token 取得 / mail: EHLO / db: SELECT 1 / server: ping
"""
import logging
import smtplib
from typing import Optional
import httpx
import psycopg2

from app.core.config import settings

logger = logging.getLogger(__name__)


class SystemChecker:

    def check_traefik(self, host: str = "localhost", port: int = 8080) -> bool:
        """Traefik API: overview + routers count > 0"""
        try:
            resp = httpx.get(f"http://{host}:{port}/api/overview", timeout=5.0)
            if resp.status_code != 200:
                return False
            data = resp.json()
            total_routers = data.get("http", {}).get("routers", {}).get("total", 0)
            return total_routers > 0
        except Exception as e:
            logger.warning(f"Traefik check failed: {e}")
            return False

    def check_keycloak(self, base_url: str, realm: str = "master") -> bool:
        """
        Step 1: GET /realms/master → 200
        Step 2: POST /realms/master/protocol/openid-connect/token with client_credentials
        """
        try:
            resp = httpx.get(f"{base_url}/realms/{realm}", timeout=8.0)
            if resp.status_code != 200:
                return False
            # Functional check: get admin token
            token_resp = httpx.post(
                f"{base_url}/realms/{settings.KEYCLOAK_ADMIN_REALM}/protocol/openid-connect/token",
                data={
                    "grant_type": "client_credentials",
                    "client_id": settings.KEYCLOAK_ADMIN_CLIENT_ID,
                    "client_secret": settings.KEYCLOAK_ADMIN_CLIENT_SECRET,
                },
                timeout=8.0,
            )
            return token_resp.status_code == 200 and "access_token" in token_resp.json()
        except Exception as e:
            logger.warning(f"Keycloak check failed ({base_url}): {e}")
            return False

    def check_smtp(self, host: str, port: int = 587) -> bool:
        """SMTP connect + EHLO (functional protocol check)"""
        try:
            with smtplib.SMTP(host, port, timeout=8) as smtp:
                smtp.ehlo()
            return True
        except Exception as e:
            logger.warning(f"SMTP check failed ({host}:{port}): {e}")
            return False

    def check_postgres(self, host: str, port: int = 5432) -> bool:
        """psycopg2 connect + SELECT 1"""
        try:
            conn = psycopg2.connect(
                host=host, port=port, dbname="postgres",
                user="admin", password="DC1qaz2wsx",
                connect_timeout=8,
            )
            cur = conn.cursor()
            cur.execute("SELECT 1")
            result = cur.fetchone()
            conn.close()
            return result == (1,)
        except Exception as e:
            logger.warning(f"PostgreSQL check failed ({host}:{port}): {e}")
            return False

    def ping_server(self, ip_address: str) -> Optional[float]:
        """
        ICMP ping, returns response time in ms or None if unreachable.
        Falls back to TCP port 22 if ping requires root privileges.
        """
        try:
            import ping3
            result = ping3.ping(ip_address, timeout=3)
            if result is not None and result is not False:
                return round(result * 1000, 2)  # convert to ms
        except PermissionError:
            # Fallback: TCP connect to port 22
            import socket
            import time
            try:
                start = time.time()
                sock = socket.create_connection((ip_address, 22), timeout=3)
                sock.close()
                return round((time.time() - start) * 1000, 2)
            except Exception:
                pass
        except Exception as e:
            logger.warning(f"Ping failed for {ip_address}: {e}")
        return None