79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
from services.runner import run_cmd
|
|
|
|
|
|
def _top_dirs_cmd(path: str, limit: int) -> list[str]:
|
|
_ = limit
|
|
return ["du", "-x", "-h", "-d", "1", path]
|
|
|
|
|
|
_SIZE_RE = re.compile(r"^\s*([0-9]+(?:\.[0-9]+)?)([KMGTP]?)(i?B?)?$", re.IGNORECASE)
|
|
|
|
|
|
def _size_to_bytes(value: str) -> float:
|
|
m = _SIZE_RE.match(value.strip())
|
|
if not m:
|
|
return -1.0
|
|
num = float(m.group(1))
|
|
unit = (m.group(2) or "").upper()
|
|
mul = {
|
|
"": 1,
|
|
"K": 1024,
|
|
"M": 1024**2,
|
|
"G": 1024**3,
|
|
"T": 1024**4,
|
|
"P": 1024**5,
|
|
}.get(unit, 1)
|
|
return num * mul
|
|
|
|
|
|
def _format_top_dirs(raw: str, limit: int) -> str:
|
|
rows: list[tuple[float, str]] = []
|
|
for line in raw.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
parts = line.split(maxsplit=1)
|
|
if len(parts) != 2:
|
|
continue
|
|
size, name = parts
|
|
rows.append((_size_to_bytes(size), f"{size}\t{name}"))
|
|
rows.sort(key=lambda x: x[0])
|
|
return "\n".join(line for _sz, line in rows[-max(1, 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():
|
|
top_out = _format_top_dirs(out, limit)
|
|
lines.append("")
|
|
lines.append("Top directories:")
|
|
lines.append(top_out)
|
|
|
|
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():
|
|
top_out2 = _format_top_dirs(out2, limit)
|
|
lines.append("")
|
|
lines.append(f"Docker dir: {docker_dir}")
|
|
lines.append(top_out2)
|
|
|
|
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():
|
|
top_out3 = _format_top_dirs(out3, limit)
|
|
lines.append("")
|
|
lines.append(f"Logs dir: {logs_dir}")
|
|
lines.append(top_out3)
|
|
|
|
return "\n".join(lines)
|