10-bootstrap/engineering-bootstrap.py

#!/usr/bin/env python3
"""Materialisiert die lokale Engineering-Installation aus einer Beschreibung.

Der Bootstrap ist idempotent: vorhandene Konfiguration wird nicht überschrieben,
sofern dies nicht ausdrücklich mit --force-env verlangt wird. Systemänderungen
und Datenbankanlage benötigen eigene Optionen.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import re
import shutil
import socket
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

RUNTIME_DIRS = (
    "config", "database", "services", "installations", "seed",
    "inbox/downloads", "inbox/failed", "inbox/processed",
    "outbox", "reports", "logs", "work",
)


@dataclass
class Finding:
    level: str
    subject: str
    message: str


@dataclass
class BootstrapResult:
    findings: list[Finding] = field(default_factory=list)
    changed: list[str] = field(default_factory=list)

    def add(self, level: str, subject: str, message: str) -> None:
        self.findings.append(Finding(level, subject, message))

    @property
    def has_errors(self) -> bool:
        return any(item.level == "ERROR" for item in self.findings)


def load_config(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise ValueError(f"Installationsbeschreibung fehlt: {path}") from exc
    except json.JSONDecodeError as exc:
        raise ValueError(f"Ungültiges JSON in {path}: {exc}") from exc

    required = ["installation", "repositories", "runtime", "languages", "seeds"]
    missing = [key for key in required if key not in data]
    if missing:
        raise ValueError(f"Pflichtbereiche fehlen: {', '.join(missing)}")
    return data


def parse_version(text: str) -> tuple[int, ...]:
    match = re.search(r"(\d+)(?:\.(\d+))?(?:\.(\d+))?", text)
    if not match:
        return ()
    return tuple(int(part or 0) for part in match.groups())


def command_version(command: str, args: list[str]) -> tuple[bool, str]:
    executable = shutil.which(command)
    if not executable:
        return False, "nicht installiert oder nicht in PATH"
    try:
        completed = subprocess.run(
            [executable, *args], capture_output=True, text=True, timeout=10, check=False
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return False, str(exc)
    output = (completed.stdout or completed.stderr).strip().splitlines()
    return completed.returncode == 0, output[0] if output else "keine Versionsausgabe"


def check_dependency(result: BootstrapResult, name: str, minimum: str | None = None) -> None:
    args = ["version"] if name == "go" else ["--version"]
    ok, version_text = command_version(name, args)
    if not ok:
        result.add("ERROR", f"dependency:{name}", version_text)
        return
    if minimum and parse_version(version_text) < parse_version(minimum):
        result.add("ERROR", f"dependency:{name}", f"{version_text}; benötigt >= {minimum}")
    else:
        result.add("OK", f"dependency:{name}", version_text)


def check_mariadb_service(result: BootstrapResult, service_name: str) -> None:
    systemctl = shutil.which("systemctl")
    if not systemctl:
        result.add("WARN", "mariadb-service", "systemctl nicht verfügbar; Dienststatus nicht prüfbar")
        return
    completed = subprocess.run(
        [systemctl, "is-active", service_name], capture_output=True, text=True, check=False
    )
    status = completed.stdout.strip() or completed.stderr.strip()
    if completed.returncode == 0:
        result.add("OK", "mariadb-service", status)
    else:
        result.add("ERROR", "mariadb-service", status or "Dienst nicht aktiv")


def check_port(result: BootstrapResult, host: str, port: int, environment: str) -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.settimeout(0.3)
        occupied = sock.connect_ex((host, port)) == 0
    if occupied:
        result.add("WARN", f"port:{environment}", f"{host}:{port} ist bereits belegt")
    else:
        result.add("OK", f"port:{environment}", f"{host}:{port} ist frei")


def env_content(config: dict[str, Any], environment: str) -> str:
    home = Path(config["installation"]["engineeringHome"])
    runtime = config["runtime"][environment]
    database = runtime["database"]
    lines = [
        f"ENGINEERING_HOME={home}",
        f"EP_ENVIRONMENT={environment}",
        f"EP_HTTP_HOST={runtime['host']}",
        f"EP_HTTP_PORT={runtime['port']}",
        f"EP_DATABASE_HOST={database['host']}",
        f"EP_DATABASE_PORT={database['port']}",
        f"EP_DATABASE_NAME={database['name']}",
        f"EP_DATABASE_USER={database['user']}",
        f"EP_DATABASE_PASSWORD={database['password']}",
        f"EP_RUNTIME_ROOT={home / config['repositories']['runtime'] / environment}",
        f"EP_DEFAULT_UI_LANGUAGE={config['languages']['default']}",
        f"EP_SUPPORTED_LANGUAGES={','.join(config['languages']['supported'])}",
        f"EP_ARTIFACT_SEED_ENABLED=true",
        f"EP_ARTIFACT_SEED_MODE={runtime['seedMode']}",
        f"EP_ARTIFACT_SEED_MANIFEST={home / config['seeds']['manifest']}",
        "EP_LOG_LEVEL=debug" if environment == "test" else "EP_LOG_LEVEL=info",
    ]
    return "\n".join(lines) + "\n"


def materialize_runtime(
    config: dict[str, Any], result: BootstrapResult, apply: bool, force_env: bool
) -> None:
    home = Path(config["installation"]["engineeringHome"])
    runtime_root = home / config["repositories"]["runtime"]
    for environment in ("test", "production"):
        env_root = runtime_root / environment
        for relative in RUNTIME_DIRS:
            path = env_root / relative
            if not path.exists():
                if apply:
                    path.mkdir(parents=True, exist_ok=True)
                    result.changed.append(str(path))
                result.add("PLAN" if not apply else "OK", f"directory:{environment}", str(path))

        env_file = env_root / "config" / "engineering-platform.env"
        if env_file.exists() and not force_env:
            result.add("OK", f"env:{environment}", f"vorhanden, unverändert: {env_file}")
        else:
            if apply:
                env_file.parent.mkdir(parents=True, exist_ok=True)
                env_file.write_text(env_content(config, environment), encoding="utf-8")
                os.chmod(env_file, 0o600)
                result.changed.append(str(env_file))
            action = "würde erzeugen" if not apply else "erzeugt"
            result.add("PLAN" if not apply else "OK", f"env:{environment}", f"{action}: {env_file}")


def check_repositories(config: dict[str, Any], result: BootstrapResult) -> None:
    home = Path(config["installation"]["engineeringHome"])
    for name, relative in config["repositories"].items():
        path = home / relative
        level = "OK" if path.exists() else "ERROR"
        result.add(level, f"repository:{name}", str(path))

    seed = home / config["seeds"]["manifest"]
    result.add("OK" if seed.exists() else "ERROR", "seed-manifest", str(seed))


def database_sql(config: dict[str, Any], environment: str) -> str:
    db = config["runtime"][environment]["database"]
    # Namen werden aus einer versionierten Beschreibung gelesen; nur konservative Zeichen sind zulässig.
    for key in ("name", "user"):
        if not re.fullmatch(r"[A-Za-z0-9_]+", db[key]):
            raise ValueError(f"Unsicherer MariaDB-Wert für {environment}.{key}: {db[key]!r}")
    password = db["password"].replace("'", "''")
    return (
        f"CREATE DATABASE IF NOT EXISTS `{db['name']}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;\n"
        f"CREATE USER IF NOT EXISTS '{db['user']}'@'localhost' IDENTIFIED BY '{password}';\n"
        f"ALTER USER '{db['user']}'@'localhost' IDENTIFIED BY '{password}';\n"
        f"GRANT ALL PRIVILEGES ON `{db['name']}`.* TO '{db['user']}'@'localhost';\n"
        "FLUSH PRIVILEGES;\n"
    )


def configure_database(config: dict[str, Any], result: BootstrapResult, apply: bool) -> None:
    mariadb = shutil.which("mariadb")
    if not mariadb:
        result.add("ERROR", "database-config", "mariadb-Client fehlt")
        return
    for environment in ("test", "production"):
        sql = database_sql(config, environment)
        if not apply:
            result.add("PLAN", f"database:{environment}", "würde Datenbank und lokalen Benutzer anlegen")
            continue
        completed = subprocess.run(
            ["sudo", mariadb], input=sql, text=True, capture_output=True, check=False
        )
        if completed.returncode == 0:
            result.add("OK", f"database:{environment}", "Datenbank und Benutzer konfiguriert")
        else:
            result.add("ERROR", f"database:{environment}", completed.stderr.strip())


def write_report(config: dict[str, Any], result: BootstrapResult, report_path: Path | None) -> Path:
    home = Path(config["installation"]["engineeringHome"])
    if report_path is None:
        report_path = home / config["installation"]["reportDirectory"] / "engineering-bootstrap-report.md"
    report_path.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "# Engineering Bootstrap Report",
        "",
        f"Zeitpunkt: {dt.datetime.now(dt.timezone.utc).isoformat()}",
        f"Engineering Home: `{home}`",
        "",
        "## Findings",
        "",
        "| Status | Bereich | Ergebnis |",
        "|---|---|---|",
    ]
    for item in result.findings:
        message = item.message.replace("|", "\\|")
        lines.append(f"| {item.level} | `{item.subject}` | {message} |")
    lines.extend(["", "## Änderungen", ""])
    if result.changed:
        lines.extend(f"- `{path}`" for path in result.changed)
    else:
        lines.append("Keine Dateien oder Verzeichnisse verändert.")
    report_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return report_path


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("config", type=Path, help="Installationsbeschreibung im JSON-Format")
    parser.add_argument("--apply", action="store_true", help="Verzeichnisse und lokale .env-Dateien erzeugen")
    parser.add_argument("--configure-database", action="store_true", help="native MariaDB-Datenbanken und Benutzer konfigurieren")
    parser.add_argument("--force-env", action="store_true", help="bestehende .env-Dateien ausdrücklich überschreiben")
    parser.add_argument("--report", type=Path, help="alternativer Reportpfad")
    args = parser.parse_args()

    try:
        config = load_config(args.config)
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2

    result = BootstrapResult()
    check_dependency(result, "go", config.get("dependencies", {}).get("go", {}).get("minimumVersion"))
    check_dependency(result, "node")
    check_dependency(result, "npm")
    check_dependency(result, "python3", config.get("dependencies", {}).get("python3", {}).get("minimumVersion"))
    check_dependency(result, "mariadb")
    check_mariadb_service(result, config.get("dependencies", {}).get("mariadbService", {}).get("serviceName", "mariadb"))
    check_repositories(config, result)
    for environment in ("test", "production"):
        runtime = config["runtime"][environment]
        check_port(result, runtime["host"], int(runtime["port"]), environment)
    materialize_runtime(config, result, args.apply, args.force_env)
    if args.configure_database:
        configure_database(config, result, args.apply)

    report = write_report(config, result, args.report)
    print(f"Bootstrap report: {report}")
    return 1 if result.has_errors else 0


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