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


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)


def definir_groupes_hote(data: dict, host: str, groups: list[str]) -> None:
    vars_for_host = variables_hote(data, host)
    if not groupes_hote(data, host):
        raise ValueError(f"Hote absent de l'inventaire: {host}")
    retirer_hote_de_tous_les_groupes(data, 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:
        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}")

    for playbook in playbooks:
        print(playbook)


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")

    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)

    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)

    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,
            )
            ecrire_inventaire(args.inventaire, data)
        elif args.command == "definir-groupes":
            definir_groupes_hote(data, args.hote, analyser_groupes(args.groupes))
            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)
    except Exception as exc:
        print(f"erreur: {exc}", file=sys.stderr)
        return 2

    return 0


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