codexpi/scripts/pi_doctor_collector.py

#!/usr/bin/env python3
"""Collect host health for Pi Doctor and upload it without privileged mutations."""

from __future__ import annotations

import argparse
import json
import os
import socket
import stat
import subprocess
import sys
import urllib.error
import urllib.request
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import psutil


def utcnow() -> str:
    return datetime.now(UTC).isoformat()


def _temperature() -> float | None:
    try:
        values = psutil.sensors_temperatures()
    except (AttributeError, OSError):
        values = {}
    for entries in values.values():
        for entry in entries:
            if entry.current is not None:
                return round(float(entry.current), 1)
    thermal = Path("/sys/class/thermal/thermal_zone0/temp")
    try:
        return round(float(thermal.read_text(encoding="utf-8").strip()) / 1000, 1)
    except (OSError, ValueError):
        return None


def _run(command: list[str], timeout: float = 5) -> subprocess.CompletedProcess[str]:
    return subprocess.run(command, capture_output=True, text=True, timeout=timeout, check=False)


def _docker_snapshot() -> dict[str, Any]:
    try:
        result = _run(["docker", "ps", "--all", "--format", "{{json .}}"])
    except (OSError, subprocess.SubprocessError) as exc:
        return {"available": False, "containers": [], "error": str(exc)[:300]}
    if result.returncode:
        return {"available": False, "containers": [], "error": result.stderr.strip()[:300]}
    containers = []
    for line in result.stdout.splitlines():
        try:
            item = json.loads(line)
        except json.JSONDecodeError:
            continue
        containers.append({"name": item.get("Names", ""), "state": item.get("State", "unknown"), "status": item.get("Status", "")})
    return {"available": True, "containers": containers}


def _mounts() -> list[dict[str, Any]]:
    rows = []
    for part in psutil.disk_partitions(all=True):
        if part.fstype.lower() not in {"cifs", "smb3"} and not part.mountpoint.startswith(("/media/", "/mnt/")):
            continue
        row: dict[str, Any] = {"path": part.mountpoint, "fstype": part.fstype, "state": "mounted"}
        try:
            row["percent"] = psutil.disk_usage(part.mountpoint).percent
        except OSError as exc:
            row.update(state="unavailable", error=str(exc)[:300])
        rows.append(row)
    return rows


def _probe(name: str, url: str | None) -> dict[str, Any]:
    if not url:
        return {"configured": False}
    try:
        request = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "CodexPi-PiDoctor/1"})
        with urllib.request.urlopen(request, timeout=4) as response:
            return {"configured": True, "available": response.status < 500, "status": response.status}
    except urllib.error.HTTPError as exc:
        return {"configured": True, "available": exc.code < 500, "status": exc.code}
    except (OSError, urllib.error.URLError) as exc:
        return {"configured": True, "available": False, "error": f"{name}: {exc}"[:300]}


def collect(codexpi_url: str | None = None, jellyfin_url: str | None = None) -> dict[str, Any]:
    cpu = psutil.cpu_percent(interval=0.2)
    memory = psutil.virtual_memory().percent
    root = psutil.disk_usage("/").percent
    return {
        "system": {
            "timestamp": utcnow(), "hostname": socket.gethostname(), "cpu_percent": cpu,
            "memory_percent": memory, "temperature_c": _temperature(),
            "boot_time": datetime.fromtimestamp(psutil.boot_time(), UTC).isoformat(),
            "kernel_error_count": 0, "previous_boot_error_count": 0, "abnormal_restart_count": 0,
        },
        "storage": {"root_percent": root, "mounts": _mounts()},
        "docker": _docker_snapshot(),
        "network": {},
        "services": {"codexpi": _probe("CodexPi", codexpi_url), "jellyfin": _probe("Jellyfin", jellyfin_url)},
    }


def detect_incidents(snapshot: dict[str, Any]) -> list[dict[str, Any]]:
    system, storage = snapshot["system"], snapshot["storage"]
    now, rows = system["timestamp"], []

    def add(key: str, severity: str, service: str, symptoms: str, details: dict[str, Any]) -> None:
        rows.append({"fingerprint": key, "source": "pi-doctor-collector", "severity": severity,
                     "service": service, "symptoms": symptoms, "details": details,
                     "first_seen": now, "last_seen": now})

    for key, value, warning, critical, label in (
        ("cpu-high", system["cpu_percent"], 90, 98, "CPU"),
        ("memory-high", system["memory_percent"], 90, 97, "RAM"),
        ("root-disk-high", storage["root_percent"], 85, 95, "Disque système"),
    ):
        if value >= warning:
            add(key, "critical" if value >= critical else "warning", label, f"Utilisation élevée : {value:.1f} %", {"percent": value})
    temperature = system.get("temperature_c")
    if temperature is not None and temperature >= 75:
        add("temperature-high", "critical" if temperature >= 85 else "warning", "Température", f"Température élevée : {temperature:.1f} °C", {"temperature_c": temperature})
    if not snapshot["docker"].get("available"):
        add("docker-unavailable", "warning", "Docker", "Docker indisponible", {"error": snapshot["docker"].get("error")})
    for mount in storage["mounts"]:
        if mount.get("state") != "mounted":
            add(f"mount:{mount['path']}", "warning", mount["path"], "Montage SMB indisponible", mount)
    for name, state in snapshot["services"].items():
        if state.get("configured") and not state.get("available"):
            add(f"service:{name}", "warning", name, "Service configuré mais indisponible", state)
    return rows


def _post(url: str, token: str, payload: dict[str, Any]) -> None:
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(url, data=body, method="POST", headers={
        "Content-Type": "application/json", "X-Pi-Doctor-Token": token,
        "User-Agent": "CodexPi-PiDoctor/1",
    })
    with urllib.request.urlopen(request, timeout=10) as response:
        if response.status >= 300:
            raise RuntimeError(f"Réponse HTTP {response.status}")


def deliver(base_url: str, token: str, snapshot: dict[str, Any]) -> None:
    base = base_url.rstrip("/")
    _post(f"{base}/api/pi-doctor/snapshots/import", token, {"snapshot": snapshot})
    incidents = detect_incidents(snapshot)
    if incidents:
        _post(f"{base}/api/pi-doctor/incidents/import", token, {"incidents": incidents})


def enqueue(spool: Path, snapshot: dict[str, Any]) -> Path:
    spool.mkdir(parents=True, exist_ok=True, mode=0o700)
    target = spool / f"{snapshot['system']['timestamp'].replace(':', '-')}-{uuid.uuid4().hex}.json"
    target.write_text(json.dumps(snapshot, ensure_ascii=False), encoding="utf-8")
    target.chmod(0o600)
    return target


def flush(spool: Path, base_url: str, token: str) -> int:
    delivered = 0
    for path in sorted(spool.glob("*.json"))[:100]:
        snapshot = json.loads(path.read_text(encoding="utf-8"))
        deliver(base_url, token, snapshot)
        path.unlink()
        delivered += 1
    return delivered


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Collecteur hôte en lecture seule pour Pi Doctor")
    parser.add_argument("--url", default=os.getenv("CODEXPI_URL", "https://localhost"))
    parser.add_argument("--token-file", type=Path, default=Path(os.getenv("PI_DOCTOR_TOKEN_FILE", "/var/lib/codexpi/pi-doctor/ingest.token")))
    parser.add_argument("--spool", type=Path, default=Path(os.getenv("PI_DOCTOR_SPOOL", "/var/lib/codexpi/pi-doctor/spool")))
    parser.add_argument("--jellyfin-url", default=os.getenv("JELLYFIN_HEALTH_URL"))
    args = parser.parse_args(argv)
    try:
        if stat.S_IMODE(args.token_file.stat().st_mode) & 0o077:
            raise ValueError("permissions du jeton trop ouvertes")
        token = args.token_file.read_text(encoding="utf-8").strip()
        if not token:
            raise ValueError("jeton vide")
    except (OSError, ValueError) as exc:
        print(f"Pi Doctor: jeton indisponible: {exc}", file=sys.stderr)
        return 2
    snapshot = collect(args.url, args.jellyfin_url)
    enqueue(args.spool, snapshot)
    try:
        flush(args.spool, args.url, token)
    except (OSError, RuntimeError, ValueError, urllib.error.URLError) as exc:
        print(f"Pi Doctor: collecte conservée pour réessai: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())