Add disk usage snapshot reports

services/disk_report.py

@@ -0,0 +1,38 @@
+import os
+from typing import Any
+
+from services.runner import run_cmd
+
+
+def _top_dirs_cmd(path: str, limit: int) -> list[str]:
+    return ["bash", "-lc", f"du -xhd1 {path} 2>/dev/null | sort -h | tail -n {limit}"]
+
+
+async def build_disk_report(cfg: dict[str, Any], mount: str, usage: int) -> str:
+    limit = int(cfg.get("disk_report", {}).get("top_dirs", 8))
+
+    lines = ["🧱 Disk report", f"💽 {mount}: {usage}%"]
+
+    rc, out = await run_cmd(_top_dirs_cmd(mount, limit), timeout=30)
+    if rc == 0 and out.strip():
+        lines.append("")
+        lines.append("Top directories:")
+        lines.append(out.strip())
+
+    docker_dir = cfg.get("disk_report", {}).get("docker_dir", "/var/lib/docker")
+    if docker_dir and os.path.exists(docker_dir):
+        rc2, out2 = await run_cmd(_top_dirs_cmd(docker_dir, limit), timeout=30)
+        if rc2 == 0 and out2.strip():
+            lines.append("")
+            lines.append(f"Docker dir: {docker_dir}")
+            lines.append(out2.strip())
+
+    logs_dir = cfg.get("disk_report", {}).get("logs_dir", "/var/log")
+    if logs_dir and os.path.exists(logs_dir):
+        rc3, out3 = await run_cmd(_top_dirs_cmd(logs_dir, limit), timeout=30)
+        if rc3 == 0 and out3.strip():
+            lines.append("")
+            lines.append(f"Logs dir: {logs_dir}")
+            lines.append(out3.strip())
+
+    return "\n".join(lines)