infra/ansible/scripts/gen-inventory-docs.py

#!/usr/bin/env python3
"""Печатает markdown-документацию по Ansible-инвентарю HomeLab в stdout.

Источник правды — `ansible-inventory -i inventory/hosts.yml --list`, поэтому вывод
всегда совпадает с тем, что реально видит Ansible. Файлы не пишутся: результат
вставляется в README/Obsidian вручную.

Запуск:
    make docs
    python3 scripts/gen-inventory-docs.py

Бинарь ansible-inventory ищется в порядке:
    1. $ANSIBLE_INVENTORY_BIN
    2. ./.venv/bin/ansible-inventory
    3. ansible-inventory из PATH
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
from pathlib import Path

ANSIBLE_DIR = Path(__file__).resolve().parent.parent
INVENTORY = ANSIBLE_DIR / "inventory" / "hosts.yml"

# Служебные группы, которые не несут смысла в документации.
HIDDEN_GROUPS = {"all", "ungrouped"}
EMPTY = "—"


def inventory_binaries() -> list[str]:
    candidates: list[str] = []
    from_env = os.environ.get("ANSIBLE_INVENTORY_BIN", "").strip()
    if from_env:
        candidates.append(from_env)
    candidates.append(str(ANSIBLE_DIR / ".venv" / "bin" / "ansible-inventory"))
    on_path = shutil.which("ansible-inventory")
    if on_path:
        candidates.append(on_path)

    unique: list[str] = []
    for candidate in candidates:
        if candidate not in unique:
            unique.append(candidate)
    return unique


def run_inventory(binary: str, env: dict) -> tuple[dict | None, str]:
    """Один запуск ansible-inventory. Возвращает (данные, описание ошибки)."""
    try:
        done = subprocess.run(
            [binary, "-i", str(INVENTORY), "--list"],
            cwd=str(ANSIBLE_DIR),
            env=env,
            # stdin закрыт, чтобы ansible не ушёл в интерактивный запрос пароля Vault.
            stdin=subprocess.DEVNULL,
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError as exc:
        return None, str(exc)

    if done.returncode != 0:
        tail = (done.stderr or done.stdout or "").strip().splitlines()
        return None, tail[-1] if tail else f"код возврата {done.returncode}"

    try:
        return json.loads(done.stdout), ""
    except json.JSONDecodeError as exc:
        return None, f"не удалось разобрать JSON ({exc})"


def load_inventory() -> dict:
    base_env = dict(os.environ)
    base_env["ANSIBLE_NOCOLOR"] = "1"
    base_env.pop("ANSIBLE_INVENTORY_BIN", None)

    # Запасной режим: host_vars/group_vars могут быть зашифрованы Ansible Vault
    # (inventory/host_vars/gyro/vault.yml). Для таблицы хостов они не нужны, поэтому
    # при отказе основного режима vars-плагины отключаются и инвентарь читается
    # только из hosts.yml.
    no_vars_env = dict(base_env)
    no_vars_env["ANSIBLE_VARS_ENABLED"] = ""

    problems: list[str] = []
    for binary in inventory_binaries():
        data, error = run_inventory(binary, base_env)
        if data is not None:
            return data
        problems.append(f"{binary}: {error}")

        data, fallback_error = run_inventory(binary, no_vars_env)
        if data is not None:
            sys.stderr.write(
                "ВНИМАНИЕ: host_vars/group_vars не прочитаны "
                f"({error}); таблица построена только по {INVENTORY.name}.\n"
            )
            return data
        problems.append(f"{binary} (без vars-плагинов): {fallback_error}")

    sys.stderr.write("ОШИБКА: не удалось прочитать инвентарь.\n")
    for problem in problems or ["ansible-inventory не найден"]:
        sys.stderr.write(f"  - {problem}\n")
    sys.stderr.write(
        "Установи зависимости (`make setup`), войди в nix-окружение "
        "или задай ANSIBLE_INVENTORY_BIN явно.\n"
    )
    raise SystemExit(1)


def parse_groups(data: dict) -> dict[str, dict[str, list[str]]]:
    groups: dict[str, dict[str, list[str]]] = {}
    for name, body in data.items():
        if name == "_meta" or not isinstance(body, dict):
            continue
        groups[name] = {
            "hosts": list(body.get("hosts", [])),
            "children": list(body.get("children", [])),
        }
    return groups


def resolve_hosts(group: str, groups: dict, seen: set[str] | None = None) -> set[str]:
    """Все хосты группы с учётом вложенных children."""
    seen = seen or set()
    if group in seen or group not in groups:
        return set()
    seen.add(group)
    hosts = set(groups[group]["hosts"])
    for child in groups[group]["children"]:
        hosts |= resolve_hosts(child, groups, seen)
    return hosts


def escape(value: object) -> str:
    text = str(value).strip()
    return text.replace("|", "\\|") if text else EMPTY


def main() -> int:
    data = load_inventory()
    hostvars = data.get("_meta", {}).get("hostvars", {})
    groups = parse_groups(data)

    resolved = {name: resolve_hosts(name, groups) for name in groups}

    all_hosts = set(hostvars)
    for members in resolved.values():
        all_hosts |= members

    host_groups: dict[str, list[str]] = {}
    for host in all_hosts:
        host_groups[host] = sorted(
            name
            for name, members in resolved.items()
            if host in members and name not in HIDDEN_GROUPS
        )

    out = sys.stdout.write
    out("## Хосты\n\n")
    out("| Хост | ansible_host | expected_lan_ip | Группы |\n")
    out("|---|---|---|---|\n")
    for host in sorted(all_hosts):
        facts = hostvars.get(host, {})
        out(
            "| `{host}` | {ansible_host} | {lan_ip} | {groups} |\n".format(
                host=host,
                ansible_host=escape(facts.get("ansible_host", "")),
                lan_ip=escape(facts.get("expected_lan_ip", "")),
                groups=", ".join(f"`{g}`" for g in host_groups[host]) or EMPTY,
            )
        )

    out("\n## Группы\n\n")
    for name in sorted(groups):
        if name in HIDDEN_GROUPS:
            continue
        members = sorted(resolved[name])
        children = sorted(groups[name]["children"])
        out(f"### `{name}` ({len(members)})\n\n")
        if children:
            out("- Вложенные группы: " + ", ".join(f"`{c}`" for c in children) + "\n")
        out("- Хосты: " + (", ".join(f"`{m}`" for m in members) or EMPTY) + "\n\n")

    out(f"_Сгенерировано из `{INVENTORY.relative_to(ANSIBLE_DIR)}` "
        "через `make docs`._\n")
    return 0


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