#!/usr/bin/env python3
"""Interface web locale pour gerer un inventaire Set-OPS."""
from __future__ import annotations
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import argparse
import json
from pathlib import Path
from urllib.parse import urlparse
import yaml
from inventory_rules import (
GROUPE_HOTES_ACTIFS,
GROUPE_HOTES_PLANIFIES,
GROUPES_ETAT_HOTE,
charger_dependances,
est_groupe_operationnel,
groupes_operationnels_connus,
)
RACINE = Path(__file__).resolve().parents[1]
INVENTAIRE_DEFAUT = RACINE / "inventories/production/hosts.yml"
DOSSIER_PLAYBOOKS_GROUPES = RACINE / "playbooks/groupes"
FICHIER_DEPENDANCES = RACINE / "docs/dependances-groupes.yml"
def charger_yaml(path: Path) -> dict:
if not path.exists():
return {"all": {"children": {}}}
with path.open("r", encoding="utf-8") as fichier:
data = yaml.safe_load(fichier) 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_yaml(path: Path, data: dict) -> None:
with path.open("w", encoding="utf-8") as fichier:
yaml.safe_dump(data, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True)
def enfants(data: dict) -> dict:
return data.setdefault("all", {}).setdefault("children", {})
def groupes_disponibles(data: dict) -> list[str]:
return groupes_operationnels_connus(enfants(data), DOSSIER_PLAYBOOKS_GROUPES)
def variables_hote(data: dict, hote: str) -> dict:
resultat: dict = {}
for groupe_data in enfants(data).values():
hosts = groupe_data.get("hosts", {})
valeurs = hosts.get(hote)
if isinstance(valeurs, dict):
resultat.update(valeurs)
return resultat
def liste_hotes(data: dict) -> list[dict]:
noms = set()
for groupe_data in enfants(data).values():
noms.update(groupe_data.get("hosts", {}))
hotes = []
for nom in sorted(noms):
groupes = []
etat = "planifie"
visible = False
for groupe, groupe_data in enfants(data).items():
if nom not in groupe_data.get("hosts", {}):
continue
if groupe == GROUPE_HOTES_ACTIFS:
etat = "actif"
visible = True
elif groupe in GROUPES_ETAT_HOTE:
visible = True
continue
elif est_groupe_operationnel(groupe):
groupes.append(groupe)
visible = True
if not visible:
continue
variables = variables_hote(data, nom)
hotes.append(
{
"nom": nom,
"etat": etat,
"adresse_ip": variables.get("ansible_host", ""),
"utilisateur_ansible": variables.get("ansible_user", "ansible"),
"vmid": variables.get("proxmox_vmid", ""),
"groupes": sorted(groupes),
}
)
return hotes
def inventaire_api(path: Path) -> dict:
data = charger_yaml(path)
dependencies = charger_dependances(FICHIER_DEPENDANCES)
return {
"inventaire": str(path.relative_to(RACINE)),
"groupes": groupes_disponibles(data),
"dependances": dependencies,
"hotes": liste_hotes(data),
}
def donnees_hote(hote: dict) -> dict:
variables: dict = {}
adresse_ip = str(hote.get("adresse_ip", "")).strip()
utilisateur = str(hote.get("utilisateur_ansible", "")).strip()
vmid = str(hote.get("vmid", "")).strip()
if adresse_ip:
variables["ansible_host"] = adresse_ip
if utilisateur:
variables["ansible_user"] = utilisateur
if vmid:
variables["proxmox_vmid"] = int(vmid) if vmid.isdigit() else vmid
return variables
def construire_inventaire(payload: dict, groupes_connus: list[str]) -> dict:
enfants_data: dict = {
"modeles_vm": {"hosts": {}},
GROUPE_HOTES_ACTIFS: {"hosts": {}},
GROUPE_HOTES_PLANIFIES: {"hosts": {}},
}
for groupe in groupes_connus:
enfants_data.setdefault(groupe, {"hosts": {}})
enfants_data.setdefault("hotes_proxmox", {"hosts": {}})
for hote in payload.get("hotes", []):
nom = str(hote.get("nom", "")).strip()
if not nom:
continue
variables = donnees_hote(hote)
groupe_etat = GROUPE_HOTES_ACTIFS if hote.get("etat") == "actif" else GROUPE_HOTES_PLANIFIES
enfants_data[groupe_etat]["hosts"][nom] = variables
for groupe in hote.get("groupes", []):
if groupe not in groupes_connus:
continue
enfants_data[groupe]["hosts"][nom] = {}
return {"all": {"children": enfants_data}}
def valider_dependances_payload(payload: dict, dependencies: dict) -> None:
hotes = payload.get("hotes", [])
active_groups: set[str] = set()
for hote in hotes:
if hote.get("etat") != "actif":
continue
active_groups.update(hote.get("groupes", []))
missing: list[str] = []
for hote in hotes:
if hote.get("etat") != "actif":
continue
nom = str(hote.get("nom", "")).strip()
for groupe in hote.get("groupes", []):
for required_group in dependencies.get(groupe, {}).get("requiert_groupes_actifs", []):
if required_group not in active_groups:
missing.append(f"{nom}: {groupe} requiert {required_group} actif")
if missing:
raise ValueError("Dependances manquantes: " + "; ".join(missing))
def valider_payload(payload: dict, groupes_connus: list[str], dependencies: dict) -> None:
noms: set[str] = set()
vmids: dict[str, str] = {}
for hote in payload.get("hotes", []):
nom = str(hote.get("nom", "")).strip()
if not nom:
raise ValueError("Un hote a un nom vide.")
if nom in noms:
raise ValueError(f"Hote en double: {nom}")
noms.add(nom)
if hote.get("etat") == "actif" and not str(hote.get("adresse_ip", "")).strip():
raise ValueError(f"Hote actif sans adresse IP: {nom}")
groupes = hote.get("groupes", [])
if not groupes:
raise ValueError(f"Aucun groupe operationnel pour: {nom}")
inconnus = sorted(set(groupes) - set(groupes_connus))
if inconnus:
raise ValueError(f"Groupes inconnus pour {nom}: {', '.join(inconnus)}")
vmid = str(hote.get("vmid", "")).strip()
if not vmid:
continue
if vmid in vmids:
raise ValueError(f"VMID en double: {vmid} pour {vmids[vmid]} et {nom}")
vmids[vmid] = nom
valider_dependances_payload(payload, dependencies)
HTML = r"""
Set-OPS - Inventaire
"""
class Gestionnaire(BaseHTTPRequestHandler):
inventaire: Path = INVENTAIRE_DEFAUT
def repondre(self, status: int, contenu: bytes, content_type: str) -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(contenu)))
self.end_headers()
self.wfile.write(contenu)
def repondre_json(self, status: int, data: dict) -> None:
self.repondre(status, json.dumps(data, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8")
def do_GET(self) -> None:
chemin = urlparse(self.path).path
try:
if chemin == "/":
self.repondre(200, HTML.encode("utf-8"), "text/html; charset=utf-8")
elif chemin == "/api/inventaire":
self.repondre_json(200, inventaire_api(self.inventaire))
else:
self.repondre_json(404, {"erreur": "Introuvable"})
except Exception as exc:
self.repondre_json(500, {"erreur": str(exc)})
def do_POST(self) -> None:
chemin = urlparse(self.path).path
try:
if chemin != "/api/inventaire":
self.repondre_json(404, {"erreur": "Introuvable"})
return
longueur = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(longueur).decode("utf-8"))
courant = charger_yaml(self.inventaire)
groupes_connus = groupes_disponibles(courant)
dependencies = charger_dependances(FICHIER_DEPENDANCES)
valider_payload(payload, groupes_connus, dependencies)
nouveau = construire_inventaire(payload, groupes_connus)
ecrire_yaml(self.inventaire, nouveau)
self.repondre_json(200, inventaire_api(self.inventaire))
except Exception as exc:
self.repondre_json(400, {"erreur": str(exc)})
def main() -> int:
parser = argparse.ArgumentParser(description="Interface web locale pour gerer un inventaire Set-OPS.")
parser.add_argument("--inventaire", type=Path, default=INVENTAIRE_DEFAUT)
parser.add_argument("--hote", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8765)
args = parser.parse_args()
Gestionnaire.inventaire = args.inventaire.resolve()
serveur = ThreadingHTTPServer((args.hote, args.port), Gestionnaire)
print(f"Interface inventaire: http://{args.hote}:{args.port}/")
print(f"Inventaire: {Gestionnaire.inventaire}")
serveur.serve_forever()
return 0
if __name__ == "__main__":
raise SystemExit(main())