life-noc/scripts/generate_services.py

#!/usr/bin/env python3

from pathlib import Path
import re
import sys
import yaml

INPUT_FILE = Path("domains.yaml")
OUTPUT_DIR = Path("icinga/services")
HOST_NAME = "life-noc"
SERVICE_TEMPLATE = "service_echeance"
DEFAULT_MOCK_STATE = "OK"
DEFAULT_MOCK_MESSAGE = "Sous contrôle"


def slugify(value: str) -> str:
    value = value.strip().lower()
    replacements = {
        "à": "a", "â": "a", "ä": "a",
        "ç": "c",
        "é": "e", "è": "e", "ê": "e", "ë": "e",
        "î": "i", "ï": "i",
        "ô": "o", "ö": "o",
        "ù": "u", "û": "u", "ü": "u",
        "ÿ": "y",
        "œ": "oe",
        "æ": "ae",
        "'": "",
        "’": "",
    }
    for old, new in replacements.items():
        value = value.replace(old, new)
    value = re.sub(r"[^a-z0-9\-]+", "-", value)
    value = re.sub(r"-{2,}", "-", value)
    return value.strip("-")


def icinga_escape(value: str) -> str:
    return value.replace("\\", "\\\\").replace('"', '\\"')


def normalize_mock_state(value: str) -> str:
    state = str(value).strip().upper()
    allowed = {"OK", "WARNING", "CRITICAL"}
    if state not in allowed:
        raise ValueError(
            f"mock_state invalide: {value}. Valeurs permises: {', '.join(sorted(allowed))}"
        )
    return state


def render_service(domain: str, item: dict) -> str:
    name = item["name"].strip()
    date = str(item["date"]).strip()
    notes = item.get("notes", "").strip()
    notes_url = item.get("notes_url", "").strip()
    action_url = item.get("action_url", "").strip()
    mock_state = normalize_mock_state(item.get("mock_state", DEFAULT_MOCK_STATE))
    mock_message = item.get("mock_message", DEFAULT_MOCK_MESSAGE).strip() or DEFAULT_MOCK_MESSAGE

    service_name = f"{domain}-{name}"

    lines = [
        f'apply Service "{icinga_escape(service_name)}" {{',
        f'  import "{SERVICE_TEMPLATE}"',
        f'  vars.date_echeance = "{icinga_escape(date)}"',
        f'  vars.mock_state = "{icinga_escape(mock_state)}"',
        f'  vars.mock_message = "{icinga_escape(mock_message)}"',
    ]

    if notes:
        lines.append(f'  notes = "{icinga_escape(notes)}"')
    if notes_url:
        lines.append(f'  notes_url = "{icinga_escape(notes_url)}"')
    if action_url:
        lines.append(f'  action_url = "{icinga_escape(action_url)}"')

    lines.append(f'  assign where host.name == "{HOST_NAME}"')
    lines.append("}")
    lines.append("")

    return "\n".join(lines)


def main() -> int:
    if not INPUT_FILE.exists():
        print(f"Erreur: fichier introuvable: {INPUT_FILE}", file=sys.stderr)
        return 1

    with INPUT_FILE.open("r", encoding="utf-8") as f:
        data = yaml.safe_load(f)

    if not isinstance(data, dict) or "domains" not in data:
        print("Erreur: le YAML doit contenir une clé racine 'domains'.", file=sys.stderr)
        return 1

    domains = data["domains"]
    if not isinstance(domains, dict):
        print("Erreur: 'domains' doit être un objet YAML.", file=sys.stderr)
        return 1

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    for existing_conf in OUTPUT_DIR.glob("*.conf"):
        existing_conf.unlink()

    generated_files = []

    for raw_domain, services in domains.items():
        domain = slugify(str(raw_domain))
        if not isinstance(services, list):
            print(f"Erreur: le domaine '{raw_domain}' doit contenir une liste.", file=sys.stderr)
            return 1

        output_file = OUTPUT_DIR / f"{domain}.conf"
        content = [
            "/*",
            f"  AUTO-GENERATED FILE - DOMAIN: {raw_domain}",
            "  Ne pas modifier manuellement.",
            "  Source: domains.yaml",
            "*/",
            "",
        ]

        for idx, item in enumerate(services, start=1):
            if not isinstance(item, dict):
                print(f"Erreur: entrée invalide dans '{raw_domain}' à la position {idx}.", file=sys.stderr)
                return 1

            missing = [key for key in ("name", "date", "notes") if key not in item]
            if missing:
                print(
                    f"Erreur: dans le domaine '{raw_domain}', entrée {idx}, champs manquants: {', '.join(missing)}",
                    file=sys.stderr,
                )
                return 1

            try:
                content.append(render_service(domain, item))
            except ValueError as exc:
                print(f"Erreur dans '{raw_domain}', entrée {idx}: {exc}", file=sys.stderr)
                return 1

        output_file.write_text("\n".join(content).rstrip() + "\n", encoding="utf-8")
        generated_files.append(str(output_file))

    print("Fichiers générés :")
    for file_path in generated_files:
        print(f"  - {file_path}")

    return 0


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