Set-OPS/scripts/applications.py

#!/usr/bin/env python3
"""Lit et valide le registre des applications Set-OPS (docs/applications.yml)."""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import yaml

from inventory_gui import charger_yaml, liste_hotes
from inventory_rules import (
    bases_de_application,
    chaine_connexion,
    charger_applications,
    charger_bases_donnees,
    charger_domaines,
    valider_applications,
)

RACINE = Path(__file__).resolve().parents[1]
FICHIER = RACINE / "docs/applications.yml"
FICHIER_BASES = RACINE / "docs/bases-donnees.yml"
FICHIER_DOMAINES = RACINE / "docs/domaines.yml"
INVENTAIRE = RACINE / "inventories/production/hosts.yml"

# Groupes 'serveurs_*' qui sont des capacites de SOCLE (sur toutes les VM),
# pas des applications a part entiere.
GROUPES_SOCLE = {"serveurs_debian", "serveurs_durcis"}


def _liste(valeur: str) -> list:
    return [v.strip() for v in (valeur or "").split(",") if v.strip()]


def ecrire(registre: dict) -> None:
    entete = (
        "# Registre des applications Set-OPS (application = entite pivot).\n"
        "# Edite par make inventaire-ui ou scripts/applications.py.\n"
        "# Une VM peut porter plusieurs applications ; une base se lie a une application.\n"
        "---\n"
    )
    with FICHIER.open("w", encoding="utf-8") as fichier:
        fichier.write(entete)
        yaml.safe_dump({"applications": registre.get("applications", {}) or {}},
                       fichier, default_flow_style=False, sort_keys=False, allow_unicode=True)


def lister(registre: dict, bases: dict) -> None:
    apps = registre.get("applications", {})
    if not apps:
        print("Aucune application declaree.")
        return
    for nom, app in apps.items():
        extras = []
        if app.get("port"):
            extras.append(f"port {app['port']}")
        if app.get("requiert"):
            extras.append(f"requiert {', '.join(app['requiert'])}")
        if app.get("expose"):
            extras.append(f"expose {', '.join(app['expose'])}")
        suffixe = ("  {" + " ; ".join(extras) + "}") if extras else ""
        print(f"{nom}  [groupe {app.get('groupe', '?')}, hote {app.get('hote', '?')}]{suffixe}")
        for entree in bases_de_application(bases, application=nom,
                                           groupe=app.get("groupe"), hote=app.get("hote")):
            base = entree["base"]
            portee = base.get("portee", "groupe")
            print(f"  DSN ({portee}/{base.get('usage', '-')}): {chaine_connexion(base, entree['serveur'])}")


def bootstrap(registre_existant: dict) -> dict:
    """(Re)genere les applications depuis les appartenances de groupes de l'inventaire.

    Un groupe 'serveurs_*' (hors socle) sur un hote = une application. Les entrees
    existantes correspondant a un couple (groupe, hote) toujours present sont
    preservees telles quelles (port / expose / requiert ...).
    """
    existant = registre_existant.get("applications") or {}
    par_cle = {(a.get("groupe"), a.get("hote")): (cle, a) for cle, a in existant.items()}
    apps: dict = {}
    for h in liste_hotes(charger_yaml(INVENTAIRE)):
        hote = h.get("nom")
        if not hote:
            continue
        for groupe in sorted(h.get("groupes", [])):
            if not groupe.startswith("serveurs_") or groupe in GROUPES_SOCLE:
                continue
            garde = par_cle.get((groupe, hote))
            if garde:
                apps[garde[0]] = garde[1]
                continue
            base = groupe[len("serveurs_"):]
            app_id = base if base not in apps else f"{base}-{hote}"
            apps[app_id] = {"groupe": groupe, "hote": hote}
    return {"applications": apps}


def main() -> int:
    parser = argparse.ArgumentParser(description="Registre des applications Set-OPS.")
    sub = parser.add_subparsers(dest="commande", required=True)
    sub.add_parser("lister", help="Affiche les applications et leurs DSN resolus.")
    sub.add_parser("verifier", help="Valide la coherence du registre.")
    sub.add_parser("bootstrap", help="(Re)genere applications.yml depuis les groupes de l'inventaire.")

    pa = sub.add_parser("ajouter", help="Ajoute ou met a jour une application.")
    pa.add_argument("--id", required=True)
    pa.add_argument("--groupe", required=True)
    pa.add_argument("--hote", required=True)
    pa.add_argument("--port", type=int)
    pa.add_argument("--requiert", default="", help="Applications dont elle depend (separees par virgule).")
    pa.add_argument("--expose", default="", help="FQDN publics qui la publient (separes par virgule).")

    pr = sub.add_parser("retirer", help="Retire une application.")
    pr.add_argument("--id", required=True)

    args = parser.parse_args()
    try:
        registre = charger_applications(FICHIER)
        domaines = charger_domaines(FICHIER_DOMAINES)
        if args.commande == "lister":
            lister(registre, charger_bases_donnees(FICHIER_BASES))
        elif args.commande == "verifier":
            valider_applications(registre, domaines)
            print("Registre des applications valide.")
        elif args.commande == "bootstrap":
            registre = bootstrap(registre)
            valider_applications(registre, domaines)
            ecrire(registre)
            print(f"docs/applications.yml genere depuis l'inventaire ({len(registre['applications'])} applications).")
        elif args.commande == "ajouter":
            app = {"groupe": args.groupe, "hote": args.hote}
            if args.port is not None:
                app["port"] = args.port
            if _liste(args.requiert):
                app["requiert"] = _liste(args.requiert)
            if _liste(args.expose):
                app["expose"] = _liste(args.expose)
            registre["applications"][args.id] = app
            valider_applications(registre, domaines)
            ecrire(registre)
            print(f"Application '{args.id}' enregistree.")
        elif args.commande == "retirer":
            registre["applications"].pop(args.id, None)
            valider_applications(registre, domaines)
            ecrire(registre)
            print(f"Application '{args.id}' retiree.")
    except Exception as exc:
        print(f"erreur: {exc}", file=sys.stderr)
        return 2
    return 0


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