"""
Perch health endpoint — one file, Python 3.8+.

Returns server + app health in the Perch Health Spec v1 format. Perch reads it on a
schedule and alerts on failures or high CPU / memory / disk.

  Standalone (any stack — WordPress, PHP, static sites, ...):
      PERCH_HEALTH_TOKEN=<secret> python3 perch_health.py          # serves :9105/health

  FastAPI:   from perch_health import fastapi_router;  app.include_router(fastapi_router())
  Flask:     from perch_health import flask_blueprint; app.register_blueprint(flask_blueprint())
  Django:    from perch_health import django_view;     path("perch-health", django_view)

Settings (environment variables):
  PERCH_HEALTH_TOKEN   required — the secret shown on your Perch monitor page
  PERCH_HEALTH_PORT    standalone port (default 9105)
  PERCH_HEALTH_DOCKER  "1" to list Docker containers (default: on in standalone mode)
  PERCH_HOST_PROC      host /proc mount when running inside Docker (e.g. /host/proc)
  PERCH_HOST_ROOT      host / mount when running inside Docker (e.g. /hostfs)
  PERCH_REDIS_URL      Redis to report, e.g. redis://:PASSWORD@127.0.0.1:6379 (comma-separated for several;
                       "off" to skip). Default: REDIS_URL, else a Redis on 127.0.0.1:6379 if one answers.

`pip install psutil` is optional but recommended: without it only Linux /proc is read
(and on macOS/Windows only load and disk are reported).
"""
import hmac
import http.client
import json
import os
import shutil
import socket
import ssl
import time
import urllib.parse

try:
    import psutil  # optional
except ImportError:
    psutil = None

PROC = os.environ.get("PERCH_HOST_PROC", "/proc")
ROOT = os.environ.get("PERCH_HOST_ROOT", "/")
DOCKER_SOCKET = "/var/run/docker.sock"
if psutil and PROC != "/proc":
    psutil.PROCFS_PATH = PROC

_checks = {}


def register_check(name):
    """Add an app check, e.g. a database ping. Return True/False, or ("warn", "message").

        @register_check("database")
        def db():
            connection.cursor().execute("SELECT 1")
            return True
    """
    def deco(fn):
        _checks[name] = fn
        return fn
    return deco


def is_authorized(authorization_header):
    """Constant-time check of the 'Authorization: Bearer <token>' header Perch sends."""
    token = os.environ.get("PERCH_HEALTH_TOKEN", "")
    if not token:
        return False
    return hmac.compare_digest((authorization_header or "").encode(), f"Bearer {token}".encode())


# ── Collectors (psutil when installed, Linux /proc otherwise) ─────────────────

def _read(path):
    try:
        with open(path) as f:
            return f.read()
    except OSError:
        return None


_last_cpu = None


def _cpu_percent():
    global _last_cpu
    if psutil:
        return psutil.cpu_percent(interval=None)
    stat = _read(f"{PROC}/stat")
    if not stat:
        return None
    values = [int(v) for v in stat.splitlines()[0].split()[1:]]
    idle, total = values[3] + values[4], sum(values)
    if _last_cpu is None:  # first call: take a short second sample
        _last_cpu = (idle, total)
        time.sleep(0.2)
        return _cpu_percent()
    d_idle, d_total = idle - _last_cpu[0], total - _last_cpu[1]
    _last_cpu = (idle, total)
    return round(100 * (1 - d_idle / d_total), 1) if d_total > 0 else None


def _memory():
    if psutil:
        m = psutil.virtual_memory()
        return {"total_bytes": m.total, "used_bytes": m.total - m.available, "percent": m.percent}
    info = _read(f"{PROC}/meminfo")
    if not info:
        return None
    kb = {l.split(":")[0]: int(l.split()[1]) * 1024 for l in info.splitlines() if len(l.split()) >= 2}
    total, available = kb.get("MemTotal"), kb.get("MemAvailable")
    if not total or available is None:
        return None
    return {"total_bytes": total, "used_bytes": total - available, "percent": round((total - available) * 100 / total, 1)}


def _disks():
    if psutil and ROOT == "/":
        disks = []
        for p in psutil.disk_partitions(all=False):
            if p.mountpoint.startswith(("/snap", "/boot/efi")) or "loop" in p.device:
                continue
            try:
                u = psutil.disk_usage(p.mountpoint)
            except OSError:
                continue
            disks.append({"mount": p.mountpoint, "total_bytes": u.total, "used_bytes": u.used, "percent": u.percent})
        if disks:
            return disks
    u = shutil.disk_usage(ROOT)
    return [{"mount": "/", "total_bytes": u.total, "used_bytes": u.used, "percent": round(u.used * 100 / u.total, 1)}]


def _network():
    if psutil:
        n = psutil.net_io_counters()
        return {"rx_bytes": n.bytes_recv, "tx_bytes": n.bytes_sent}
    dev = _read(f"{PROC}/net/dev")
    if not dev:
        return None
    rx = tx = 0
    for line in dev.splitlines()[2:]:
        iface, data = line.split(":", 1)
        if iface.strip() == "lo":
            continue
        fields = data.split()
        rx, tx = rx + int(fields[0]), tx + int(fields[8])
    return {"rx_bytes": rx, "tx_bytes": tx}


def _uptime():
    if psutil:
        return int(time.time() - psutil.boot_time())
    up = _read(f"{PROC}/uptime")
    return int(float(up.split()[0])) if up else None


def _processes():
    if psutil:
        procs = []
        for p in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
            procs.append(p.info)
        top = sorted(procs, key=lambda p: p.get("cpu_percent") or 0, reverse=True)[:10]
        return {"total": len(procs), "top": [
            {"pid": p["pid"], "name": p["name"], "cpu_percent": round(p.get("cpu_percent") or 0, 1),
             "memory_percent": round(p.get("memory_percent") or 0, 1)} for p in top]}
    try:
        return {"total": sum(1 for d in os.listdir(PROC) if d.isdigit()), "top": []}
    except OSError:
        return None


class _UnixHTTPConnection(http.client.HTTPConnection):
    def __init__(self, path):
        super().__init__("localhost", timeout=3)
        self._path = path

    def connect(self):
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.settimeout(3)
        self.sock.connect(self._path)


def _containers():
    """Name, image and state of every container (Docker socket must be mounted, read-only is fine)."""
    if not os.path.exists(DOCKER_SOCKET):
        return []
    try:
        conn = _UnixHTTPConnection(DOCKER_SOCKET)
        conn.request("GET", "/containers/json?all=1")
        items = json.loads(conn.getresponse().read())
        conn.close()
    except (OSError, ValueError):
        return []
    return [{"name": (c.get("Names") or ["?"])[0].lstrip("/"), "image": c.get("Image"), "state": c.get("State")}
            for c in items[:50]]


def _redis_call(sock, reader, *args):
    """One Redis command (RESP) → its reply; a Redis error raises RuntimeError."""
    parts = [a.encode() for a in args]
    sock.sendall(b"*%d\r\n" % len(parts) + b"".join(b"$%d\r\n%s\r\n" % (len(a), a) for a in parts))
    line = reader.readline()
    if not line:
        raise ConnectionError("connection closed")
    kind, rest = line[:1], line[1:].strip()
    if kind == b"-":
        raise RuntimeError(rest.decode(errors="replace"))
    if kind == b"$":
        size = int(rest)
        return None if size < 0 else reader.read(size + 2)[:-2].decode(errors="replace")
    return rest.decode(errors="replace")


def _redis_one(url, explicit):
    """INFO of one Redis. A Redis we only guessed (no URL given) that isn't there is left out."""
    u = urllib.parse.urlsplit(url if "://" in url else "redis://" + url)
    host, port = u.hostname or "127.0.0.1", u.port or 6379
    user, password = urllib.parse.unquote(u.username or ""), urllib.parse.unquote(u.password or "")
    entry = {"name": f"{host}:{port}"}
    started = time.perf_counter()
    try:
        sock = socket.create_connection((host, port), timeout=2)
        if u.scheme == "rediss":
            sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)
        with sock, sock.makefile("rb") as reader:
            if password:
                _redis_call(sock, reader, *(["AUTH", user, password] if user else ["AUTH", password]))
            info = _redis_call(sock, reader, "INFO") or ""
    except RuntimeError as e:  # Redis answered with an error
        message = str(e)[:200]
        if any(w in message for w in ("NOAUTH", "WRONGPASS", "invalid password", "uthentication required")):
            message = ("password rejected" if password
                       else f"needs a password: set PERCH_REDIS_URL=redis://:PASSWORD@{host}:{port}")
            return {**entry, "status": "fail" if explicit or password else "warn", "message": message}
        return {**entry, "status": "fail", "message": message}
    except (OSError, ValueError) as e:
        return {**entry, "status": "fail", "message": str(e)[:200] or "can't connect"} if explicit else None
    fields = dict(line.split(":", 1) for line in info.splitlines() if ":" in line and not line.startswith("#"))

    def num(key):
        try:
            return int(float(fields[key]))
        except (KeyError, ValueError):
            return None
    role = fields.get("role")
    return {**entry, "status": "pass", "latency_ms": round((time.perf_counter() - started) * 1000, 1),
            "version": fields.get("valkey_version") or fields.get("redis_version"),
            "role": {"master": "primary", "slave": "replica"}.get(role, role),
            "used_memory_bytes": num("used_memory"), "maxmemory_bytes": num("maxmemory"),
            "connected_clients": num("connected_clients"), "evicted_keys": num("evicted_keys")}


def _redis():
    setting = os.environ.get("PERCH_REDIS_URL") or os.environ.get("REDIS_URL") or ""
    if setting.strip().lower() == "off":
        return []
    urls = [u.strip() for u in setting.split(",") if u.strip()]
    found = [_redis_one(u, explicit=True) for u in urls] if urls else [_redis_one("redis://127.0.0.1:6379", explicit=False)]
    return [r for r in found if r]


def collect(include_containers=None):
    """Build the Perch health document."""
    if include_containers is None:
        include_containers = os.environ.get("PERCH_HEALTH_DOCKER") == "1"

    status, checks = "pass", {}
    for name, fn in _checks.items():
        started = time.perf_counter()
        try:
            result = fn()
            check_status, message = (result if isinstance(result, tuple) else ("pass" if result else "fail", ""))
        except Exception as e:  # a failing check must not break the endpoint
            check_status, message = "fail", str(e)[:200]
        checks[name] = {"status": check_status, "latency_ms": round((time.perf_counter() - started) * 1000, 1), "message": message}
        if check_status == "fail":
            status = "fail"
        elif check_status == "warn" and status == "pass":
            status = "warn"

    redis = _redis()
    if any(r["status"] == "fail" for r in redis):
        status = "fail"
    elif any(r["status"] == "warn" for r in redis) and status == "pass":
        status = "warn"

    load = list(os.getloadavg()) if hasattr(os, "getloadavg") else None
    cpu = {"percent": _cpu_percent(), "cores": os.cpu_count(), "load": [round(v, 2) for v in load] if load else None}
    doc = {
        "status": status,
        "checks": checks,
        "system": {
            "hostname": socket.gethostname(),
            "os": os.uname().sysname if hasattr(os, "uname") else os.name,
            "uptime_seconds": _uptime(),
            "cpu": {k: v for k, v in cpu.items() if v is not None},
            "memory": _memory(),
            "disks": _disks(),
            "network": _network(),
        },
        "processes": _processes(),
    }
    if redis:
        doc["redis"] = redis
    if include_containers:
        doc["containers"] = _containers()
    return doc


if psutil:
    psutil.cpu_percent(interval=None)  # prime: the next call returns usage since now


# ── Framework adapters ────────────────────────────────────────────────────────

def fastapi_router(path="/perch-health"):
    from fastapi import APIRouter, Request
    from fastapi.responses import JSONResponse

    router = APIRouter()

    @router.get(path, include_in_schema=False)
    def perch_health(request: Request):
        if not is_authorized(request.headers.get("authorization")):
            return JSONResponse({"error": "unauthorized"}, status_code=401)
        doc = collect()
        return JSONResponse(doc, status_code=503 if doc["status"] == "fail" else 200)

    return router


def flask_blueprint(path="/perch-health"):
    from flask import Blueprint, jsonify, request

    bp = Blueprint("perch_health", __name__)

    @bp.get(path)
    def perch_health():
        if not is_authorized(request.headers.get("Authorization")):
            return jsonify(error="unauthorized"), 401
        doc = collect()
        return jsonify(doc), (503 if doc["status"] == "fail" else 200)

    return bp


def django_view(request):
    from django.http import JsonResponse

    if not is_authorized(request.headers.get("Authorization")):
        return JsonResponse({"error": "unauthorized"}, status=401)
    doc = collect()
    return JsonResponse(doc, status=503 if doc["status"] == "fail" else 200)


# ── Standalone server ─────────────────────────────────────────────────────────

def serve(host="0.0.0.0", port=None):
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

    port = int(port or os.environ.get("PERCH_HEALTH_PORT", 9105))
    if not os.environ.get("PERCH_HEALTH_TOKEN"):
        raise SystemExit("Set PERCH_HEALTH_TOKEN to the secret shown on your Perch monitor page")

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path.split("?")[0] not in ("/", "/health"):
                return self._send(404, {"error": "not found"})
            if not is_authorized(self.headers.get("Authorization")):
                return self._send(401, {"error": "unauthorized"})
            doc = collect(include_containers=os.environ.get("PERCH_HEALTH_DOCKER", "1") == "1")
            self._send(503 if doc["status"] == "fail" else 200, doc)

        def _send(self, code, body):
            data = json.dumps(body).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def log_message(self, *args):  # quiet
            pass

    print(f"Perch health endpoint on http://{host}:{port}/health")
    ThreadingHTTPServer((host, port), Handler).serve_forever()


if __name__ == "__main__":
    serve()