Set-OPS/scripts/inventory_host.py

#!/usr/bin/env python3
"""Gere les hotes d'un inventaire Ansible statique Set-OPS."""

from __future__ import annotations

import argparse
from pathlib import Path
import sys

import yaml


GROUPES_OPERATIONNELS_PREFIXES = ("serveurs_", "clients_")
GROUPE_HOTES_ACTIFS = "hotes_actifs"
GROUPE_HOTES_PLANIFIES = "hotes_planifies"
GROUPES_ETAT_HOTE = {GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES}


def charger_inventaire(path: Path) -> dict:
    if not path.exists():
        return {"all": {"children": {}}}

    with path.open("r", encoding="utf-8") as inventory_file:
        data = yaml.safe_load(inventory_file) or {}

    if not isinstance(data, dict):
        raise ValueError(f"{path} ne contient pas une table YAML.")

    data.setdefault("all", {})
    data["all"].setdefault("children", {})
    return data


def ecrire_inventaire(path: Path, data: dict) -> None:
    with path.open("w", encoding="utf-8") as inventory_file:
        yaml.safe_dump(
            data,
            inventory_file,
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
        )


def enfants(data: dict) -> dict:
    return data.setdefault("all", {}).setdefault("children", {})


def assurer_groupe(data: dict, group: str) -> dict:
    group_data = enfants(data).setdefault(group, {})
    group_data.setdefault("hosts", {})
    return group_data


def variables_hote(data: dict, host: str) -> dict:
    merged: dict = {}
    for group_data in enfants(data).values():
        hosts = group_data.get("hosts", {})
        if host in hosts and isinstance(hosts[host], dict):
            merged.update(hosts[host])
    return merged


def groupes_hote(data: dict, host: str) -> list[str]:
    groups: list[str] = []
    for group, group_data in enfants(data).items():
        if host in group_data.get("hosts", {}):
            groups.append(group)
    return groups


def retirer_hote_de_tous_les_groupes(data: dict, host: str) -> None:
    for group_data in enfants(data).values():
        group_data.get("hosts", {}).pop(host, None)


def ajouter_hote(data: dict, host: str, groups: list[str], ip: str | None, ansible_user: str | None) -> None:
    vars_for_host = variables_hote(data, host)
    if ip:
        vars_for_host["ansible_host"] = ip
    if ansible_user:
        vars_for_host["ansible_user"] = ansible_user

    for group in groups:
        assurer_groupe(data, group)["hosts"][host] = dict(vars_for_host)

    if ip:
        assurer_groupe(data, GROUPE_HOTES_ACTIFS)["hosts"][host] = dict(vars_for_host)
        assurer_groupe(data, GROUPE_HOTES_PLANIFIES)["hosts"].pop(host, None)


def definir_groupes_hote(data: dict, host: str, groups: list[str]) -> None:
    vars_for_host = variables_hote(data, host)
    current_groups = groupes_hote(data, host)
    if not current_groups:
        raise ValueError(f"Hote absent de l'inventaire: {host}")
    retirer_hote_de_tous_les_groupes(data, host)
    for group in current_groups:
        if group in GROUPES_ETAT_HOTE:
            assurer_groupe(data, group)["hosts"][host] = dict(vars_for_host)
    for group in groups:
        assurer_groupe(data, group)["hosts"][host] = dict(vars_for_host)


def analyser_groupes(raw_groups: str) -> list[str]:
    groups = [group.strip() for group in raw_groups.replace(",", " ").split() if group.strip()]
    if not groups:
        raise ValueError("Au moins un groupe est requis.")
    return groups


def afficher_hote(data: dict, host: str) -> None:
    groups = groupes_hote(data, host)
    if not groups:
        raise ValueError(f"Hote absent de l'inventaire: {host}")

    yaml.safe_dump(
        {
            "hote": host,
            "groupes": groups,
            "variables": variables_hote(data, host),
        },
        sys.stdout,
        default_flow_style=False,
        sort_keys=False,
        allow_unicode=True,
    )


def afficher_playbooks_hote(data: dict, host: str, playbook_dir: Path) -> None:
    groups = groupes_hote(data, host)
    if not groups:
        raise ValueError(f"Hote absent de l'inventaire: {host}")

    missing_playbooks: list[str] = []
    playbooks: list[Path] = []

    for group in groups:
        if not group.startswith(GROUPES_OPERATIONNELS_PREFIXES):
            continue
        playbook = playbook_dir / f"{group}.yml"
        if playbook.exists():
            playbooks.append(playbook)
        else:
            missing_playbooks.append(group)

    if missing_playbooks:
        missing = ", ".join(missing_playbooks)
        raise ValueError(f"Playbook de groupe manquant pour: {missing}")

    if not playbooks:
        raise ValueError(f"Aucun playbook de groupe opérationnel pour: {host}")

    for playbook in playbooks:
        print(playbook)


def verifier_hote_actif(data: dict, host: str) -> None:
    hosts = enfants(data).get(GROUPE_HOTES_ACTIFS, {}).get("hosts", {})
    if host not in hosts:
        raise ValueError(f"Hote non actif dans l'inventaire: {host}")


def verifier_playbooks_groupes(data: dict, playbook_dir: Path) -> None:
    missing_playbooks: list[str] = []

    for group, group_data in enfants(data).items():
        hosts = group_data.get("hosts", {})
        if not hosts:
            continue
        if not group.startswith(GROUPES_OPERATIONNELS_PREFIXES):
            continue
        playbook = playbook_dir / f"{group}.yml"
        if not playbook.exists():
            missing_playbooks.append(group)

    if missing_playbooks:
        missing = ", ".join(sorted(missing_playbooks))
        raise ValueError(f"Playbook de groupe manquant pour: {missing}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Gere les hotes d'un inventaire Ansible statique Set-OPS.")
    parser.add_argument("--inventaire", required=True, type=Path)

    subparsers = parser.add_subparsers(dest="command", required=True)

    add_parser = subparsers.add_parser("ajouter", help="Ajoute ou met a jour un hote dans un ou plusieurs groupes.")
    add_parser.add_argument("--hote", required=True)
    add_parser.add_argument("--groupes", required=True)
    add_parser.add_argument("--adresse-ip")
    add_parser.add_argument("--utilisateur-ansible")
    add_parser.add_argument("--dossier-playbooks", type=Path)

    groups_parser = subparsers.add_parser("definir-groupes", help="Remplace l'appartenance aux groupes pour un hote.")
    groups_parser.add_argument("--hote", required=True)
    groups_parser.add_argument("--groupes", required=True)
    groups_parser.add_argument("--dossier-playbooks", type=Path)

    show_parser = subparsers.add_parser("afficher", help="Affiche les groupes et variables d'un hote.")
    show_parser.add_argument("--hote", required=True)

    playbooks_parser = subparsers.add_parser("playbooks", help="Affiche les playbooks de groupes pour un hote.")
    playbooks_parser.add_argument("--hote", required=True)
    playbooks_parser.add_argument("--dossier-playbooks", required=True, type=Path)

    active_parser = subparsers.add_parser("verifier-actif", help="Verifie qu'un hote est actif.")
    active_parser.add_argument("--hote", required=True)

    verify_parser = subparsers.add_parser(
        "verifier-playbooks",
        help="Verifie que les groupes operationnels avec hotes ont un playbook homonyme.",
    )
    verify_parser.add_argument("--dossier-playbooks", required=True, type=Path)

    args = parser.parse_args()

    try:
        data = charger_inventaire(args.inventaire)

        if args.command == "ajouter":
            ajouter_hote(
                data,
                args.hote,
                analyser_groupes(args.groupes),
                args.adresse_ip,
                args.utilisateur_ansible,
            )
            if args.dossier_playbooks:
                verifier_playbooks_groupes(data, args.dossier_playbooks)
            ecrire_inventaire(args.inventaire, data)
        elif args.command == "definir-groupes":
            definir_groupes_hote(data, args.hote, analyser_groupes(args.groupes))
            if args.dossier_playbooks:
                verifier_playbooks_groupes(data, args.dossier_playbooks)
            ecrire_inventaire(args.inventaire, data)
        elif args.command == "afficher":
            afficher_hote(data, args.hote)
        elif args.command == "playbooks":
            afficher_playbooks_hote(data, args.hote, args.dossier_playbooks)
        elif args.command == "verifier-actif":
            verifier_hote_actif(data, args.hote)
        elif args.command == "verifier-playbooks":
            verifier_playbooks_groupes(data, args.dossier_playbooks)
    except Exception as exc:
        print(f"erreur: {exc}", file=sys.stderr)
        return 2

    return 0


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