#!/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())