#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright (C) 2026 Christian Wirtz <doc@snowheaven.de>
# License: GPL-2.0-only, see LICENSE in the repository root.
"""Checkmk agent plugin: observe_directory

Scans the directories listed in the config file for their immediate subdirectories and
measures, for each subdirectory *and* for the watched directory itself, the total size in
bytes and the total number of files (both recursive). The result is written to the agent
output as a single section; growth rate and thresholds are computed and evaluated
server-side by the matching check plugin, so this script only ever reports the current,
absolute state -- it keeps no history of its own.

Deployment
----------
Automatically, via the Agent Bakery: enable the "Observe directory" Setup rule
(``rulesets.observe_directory.rule_spec_agent_config_observe_directory``), which deploys
this file plus a matching observe_directory.cfg (see ../bakery/observe_directory.py).
This requires the Checkmk Enterprise/Cloud/MSP edition (Agent Bakery is not available on
Raw) -- on Raw, deploy this file to the host's
``local/share/check_mk/agents/plugins/`` manually instead and hand-write
``$MK_CONFDIR/observe_directory.cfg``; see README.md for the exact format.

Dependency-free (stdlib only) since it runs unmodified on every monitored host.
"""
from __future__ import annotations

import fnmatch
import json
import os
import signal
import sys
from pathlib import Path

SECTION_NAME = "observe_directory"
CONFIG_CANDIDATES = (
    Path(os.environ.get("MK_CONFDIR", "/etc/check_mk")) / "observe_directory.cfg",
    Path("/etc/check_mk/observe_directory.cfg"),
)
DEFAULT_TIMEOUT = 20
SELF_ITEM = "."


class _Timeout(Exception):
    """Raised by the SIGALRM handler to abort a scan that runs too long."""


def _alarm_handler(signum, frame):  # noqa: ARG001 - required signal handler signature
    raise _Timeout()


def load_config() -> dict | None:
    for candidate in CONFIG_CANDIDATES:
        if not candidate.is_file():
            continue
        try:
            with candidate.open(encoding="utf-8") as handle:
                return json.load(handle)
        except (OSError, json.JSONDecodeError) as exc:
            sys.stderr.write(f"observe_directory: cannot read {candidate}: {exc}\n")
            return None
    return None


def iter_subdirectory_names(base: Path, include_hidden: bool, exclude: list[str]):
    try:
        entries = sorted(os.scandir(base), key=lambda e: e.name)
    except OSError as exc:
        sys.stderr.write(f"observe_directory: cannot list {base}: {exc}\n")
        return
    for entry in entries:
        try:
            if not entry.is_dir(follow_symlinks=False):
                continue
        except OSError:
            continue
        if not include_hidden and entry.name.startswith("."):
            continue
        if any(fnmatch.fnmatch(entry.name, pattern) for pattern in exclude):
            continue
        yield entry.name


def measure(path: Path) -> tuple[int, int, bool]:
    """Return (size_bytes, file_count, had_errors) for everything under path,
    recursively. Symlinks are not followed (their own size is counted, their
    targets are not)."""
    total_size = 0
    file_count = 0
    had_errors = False

    def onerror(_exc: OSError) -> None:
        nonlocal had_errors
        had_errors = True

    for root, _dirs, files in os.walk(path, onerror=onerror, followlinks=False):
        for name in files:
            file_count += 1
            try:
                total_size += os.lstat(os.path.join(root, name)).st_size
            except OSError:
                had_errors = True
    return total_size, file_count, had_errors


def scan_directory(entry: dict) -> list[str]:
    lines: list[str] = []
    base = Path(entry["path"])
    exclude = entry.get("exclude", [])
    include_hidden = entry.get("include_hidden", False)

    if not base.is_dir():
        lines.append(f"{base}\t{SELF_ITEM}\tnotfound\t0\t0\t0")
        return lines

    self_size, self_count, self_err = measure(base)
    lines.append(f"{base}\t{SELF_ITEM}\tok\t{self_size}\t{self_count}\t{int(self_err)}")

    for name in iter_subdirectory_names(base, include_hidden, exclude):
        sub_size, sub_count, sub_err = measure(base / name)
        lines.append(f"{base}\t{name}\tok\t{sub_size}\t{sub_count}\t{int(sub_err)}")

    return lines


def main() -> int:
    config = load_config()
    if not config or not config.get("directories"):
        # Not configured (yet) -- stay silent rather than spamming stderr on
        # every agent run.
        return 0

    timeout = int(config.get("timeout", DEFAULT_TIMEOUT))
    output = [f"<<<{SECTION_NAME}:sep(9)>>>"]

    have_alarm = hasattr(signal, "SIGALRM")
    if have_alarm:
        signal.signal(signal.SIGALRM, _alarm_handler)
        signal.alarm(timeout)

    try:
        for entry in config["directories"]:
            output.extend(scan_directory(entry))
    except _Timeout:
        sys.stderr.write(f"observe_directory: aborted after {timeout}s timeout\n")
    finally:
        if have_alarm:
            signal.alarm(0)

    print("\n".join(output))
    return 0


if __name__ == "__main__":
    sys.exit(main())
