Docker-Update/docker-update

#!/usr/bin/env python3
"""
docker-update  —  Docker Compose stack manager for sammons-server

Usage:
  docker-update          Update all stacks
  docker-update log      Browse past update logs (fzf or plain-text fallback)
  docker-update debug    Dry-run: show what would happen, no changes made
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import textwrap
from datetime import datetime
from pathlib import Path
from typing import Any

# ── optional rich ──────────────────────────────────────────────────────────
try:
    from rich.console import Console
    from rich.progress import (
        BarColumn,
        Progress,
        SpinnerColumn,
        TaskProgressColumn,
        TextColumn,
    )
    from rich.table import Table
    from rich.markup import escape as rich_escape

    HAS_RICH = True
    console = Console()
except ImportError:
    HAS_RICH = False
    console = None  # type: ignore[assignment]

# ── constants ──────────────────────────────────────────────────────────────
LOG_DIR    = Path("/var/log/docker-update")
FZF_HELPER = "docker-update-fzf"

SUCCEEDED      = "succeeded"
FAILED_PULL    = "failed_pull"
FAILED_STOP    = "failed_stop"
FAILED_RESTART = "failed_restart"

OUTCOME_ORDER = [SUCCEEDED, FAILED_PULL, FAILED_STOP, FAILED_RESTART]

OUTCOME_LABELS: dict[str, str] = {
    SUCCEEDED:      "Succeeded",
    FAILED_PULL:    "Failed to pull (restarted)",
    FAILED_STOP:    "Failed to stop",
    FAILED_RESTART: "Failed to restart",
}

OUTCOME_ICONS: dict[str, str] = {
    SUCCEEDED:      "✓",
    FAILED_PULL:    "⚠",
    FAILED_STOP:    "✗",
    FAILED_RESTART: "✗",
}

OUTCOME_COLORS: dict[str, str] = {
    SUCCEEDED:      "green",
    FAILED_PULL:    "yellow",
    FAILED_STOP:    "red",
    FAILED_RESTART: "red",
}


# ── subprocess helper ──────────────────────────────────────────────────────

def _run(cmd: list[str], cwd: str | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        cmd, cwd=cwd, capture_output=True, text=True
    )


# ── project discovery ──────────────────────────────────────────────────────

def discover_projects() -> list[dict[str, Any]]:
    """Return sorted list of compose projects found on this host."""
    try:
        r = _run(["docker", "compose", "ls", "--all", "--format", "json"])
    except FileNotFoundError:
        sys.exit("error: 'docker' not found in PATH")

    if r.returncode != 0:
        sys.exit(f"error: docker compose ls failed:\n{r.stderr.strip()}")

    try:
        raw: list[dict[str, Any]] = json.loads(r.stdout or "[]")
    except json.JSONDecodeError:
        sys.exit(f"error: could not parse docker compose ls output:\n{r.stdout[:200]}")

    projects: list[dict[str, Any]] = []
    for item in raw:
        name   = item.get("Name", "").strip()
        config = item.get("ConfigFiles", "").strip()
        if not name or not config:
            continue
        workdir = str(Path(config.split(",")[0]).parent)
        projects.append({
            "name":    name,
            "workdir": workdir,
            "is_git":  Path(workdir, ".git").is_dir(),
            "status":  item.get("Status", ""),
        })

    return sorted(projects, key=lambda p: p["name"].lower())


# ── update logic ───────────────────────────────────────────────────────────

def _container_logs(workdir: str) -> str:
    r = _run(["docker", "compose", "logs", "--tail=100", "--no-color"], cwd=workdir)
    return (r.stdout or r.stderr or "(no logs available)").strip()


def update_project_git(project: dict[str, Any]) -> dict[str, Any]:
    workdir   = project["workdir"]
    log_lines: list[str] = []

    r = _run(["git", "pull"], cwd=workdir)
    log_lines.append(f"=== git pull (exit {r.returncode}) ===\n{r.stdout}{r.stderr}")
    pull_ok = r.returncode == 0

    r = _run(["docker", "compose", "up", "-d", "--build"], cwd=workdir)
    log_lines.append(
        f"=== docker compose up -d --build (exit {r.returncode}) ===\n{r.stdout}{r.stderr}"
    )

    if r.returncode != 0:
        log_lines.append(
            f"\nContainer logs ({project['name']}):\n{_container_logs(workdir)}"
        )
        outcome = FAILED_RESTART
    elif not pull_ok:
        outcome = FAILED_PULL
    else:
        outcome = SUCCEEDED

    return {**project, "outcome": outcome, "log_lines": log_lines}


def update_project_registry(project: dict[str, Any]) -> dict[str, Any]:
    workdir   = project["workdir"]
    log_lines: list[str] = []

    r = _run(["docker", "compose", "down"], cwd=workdir)
    log_lines.append(f"=== docker compose down (exit {r.returncode}) ===\n{r.stdout}{r.stderr}")
    if r.returncode != 0:
        log_lines.append(f"\nStep 'stop' failed (exit {r.returncode}):")
        log_lines.append(r.stderr)
        return {**project, "outcome": FAILED_STOP, "log_lines": log_lines}

    r = _run(["docker", "compose", "pull"], cwd=workdir)
    log_lines.append(f"=== docker compose pull (exit {r.returncode}) ===\n{r.stdout}{r.stderr}")
    pull_ok = r.returncode == 0
    if not pull_ok:
        log_lines.append(f"\nStep 'pull' failed (exit {r.returncode}) — restarting with existing image:")
        log_lines.append(r.stderr)

    r = _run(["docker", "compose", "up", "-d"], cwd=workdir)
    log_lines.append(
        f"=== docker compose up -d (exit {r.returncode}) ===\n{r.stdout}{r.stderr}"
    )

    if r.returncode != 0:
        log_lines.append(
            f"\nContainer logs ({project['name']}):\n{_container_logs(workdir)}"
        )
        outcome = FAILED_RESTART
    elif not pull_ok:
        outcome = FAILED_PULL
    else:
        outcome = SUCCEEDED

    return {**project, "outcome": outcome, "log_lines": log_lines}


# ── rich runner ────────────────────────────────────────────────────────────

def _run_update_rich(projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []

    with Progress(
        SpinnerColumn(),
        TextColumn("[bold]{task.description:<32}"),
        BarColumn(),
        TaskProgressColumn(),
        transient=False,
        console=console,
    ) as prog:
        task = prog.add_task("Updating stacks…", total=len(projects))
        for project in projects:
            prog.update(task, description=f"  {project['name']}")
            result = (
                update_project_git(project)
                if project["is_git"]
                else update_project_registry(project)
            )
            results.append(result)
            prog.advance(task)
            color = OUTCOME_COLORS[result["outcome"]]
            icon  = OUTCOME_ICONS[result["outcome"]]
            prog.console.print(
                f"  [{color}]{icon}[/{color}]  {rich_escape(result['name'])}"
            )

    return results


# ── plain runner ───────────────────────────────────────────────────────────

def _run_update_plain(projects: list[dict[str, Any]]) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    total = len(projects)

    for i, project in enumerate(projects, 1):
        print(f"[{i}/{total}] Updating {project['name']}…", flush=True)
        result = (
            update_project_git(project)
            if project["is_git"]
            else update_project_registry(project)
        )
        results.append(result)
        icon  = OUTCOME_ICONS[result["outcome"]]
        label = OUTCOME_LABELS[result["outcome"]]
        print(f"  {icon}  {result['name']}  ({label})")

    return results


# ── summary printers ───────────────────────────────────────────────────────

def _group_by_outcome(
    results: list[dict[str, Any]],
) -> dict[str, list[dict[str, Any]]]:
    groups: dict[str, list[dict[str, Any]]] = {k: [] for k in OUTCOME_ORDER}
    for r in results:
        groups[r["outcome"]].append(r)
    return groups


def _print_summary_rich(groups: dict[str, list[dict[str, Any]]]) -> None:
    console.print()
    console.rule("[bold]Summary")
    console.print()
    parts: list[str] = []
    if groups[SUCCEEDED]:
        parts.append(f"[green]✓ {len(groups[SUCCEEDED])} succeeded[/green]")
    if groups[FAILED_PULL]:
        parts.append(f"[yellow]⚠ {len(groups[FAILED_PULL])} failed to pull[/yellow]")
    if groups[FAILED_STOP]:
        parts.append(f"[red]✗ {len(groups[FAILED_STOP])} failed to stop[/red]")
    if groups[FAILED_RESTART]:
        parts.append(f"[red]✗ {len(groups[FAILED_RESTART])} failed to restart[/red]")
    console.print("  " + "   ".join(parts))
    console.print()


def _print_summary_plain(groups: dict[str, list[dict[str, Any]]]) -> None:
    print()
    print("=" * 60)
    for outcome in OUTCOME_ORDER:
        items = groups[outcome]
        if items:
            icon  = OUTCOME_ICONS[outcome]
            label = OUTCOME_LABELS[outcome]
            print(f"{icon} {label}: {len(items)}")
            for r in items:
                print(f"    - {r['name']}")
    print("=" * 60)


# ── log writers ────────────────────────────────────────────────────────────

def _write_text_log(
    run_id: str,
    results: list[dict[str, Any]],
    path: Path,
) -> None:
    groups = _group_by_outcome(results)
    lines: list[str] = [
        f"docker-update run: {run_id}",
        "=" * 60,
        "",
    ]
    for outcome in OUTCOME_ORDER:
        items = groups[outcome]
        if items:
            icon  = OUTCOME_ICONS[outcome]
            label = OUTCOME_LABELS[outcome]
            lines.append(f"{icon} {label}: {len(items)}")
            for r in items:
                lines.append(f"    - {r['name']}")

    failures = [r for r in results if r["outcome"] != SUCCEEDED]
    if failures:
        lines += ["", "=" * 60, "FAILURE DETAILS", "=" * 60]
        for r in failures:
            sep = "─" * 60
            lines += [
                "",
                sep,
                f"  {r['name']}  [{OUTCOME_LABELS[r['outcome']]}]",
                sep,
                "",
            ]
            for chunk in r.get("log_lines", []):
                lines.append(chunk)

    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def _write_json_log(
    run_id: str,
    results: list[dict[str, Any]],
    path: Path,
) -> None:
    data = {
        "run_id": run_id,
        "projects": [
            {
                "name":    r["name"],
                "workdir": r["workdir"],
                "is_git":  r["is_git"],
                "outcome": r["outcome"],
            }
            for r in results
        ],
    }
    path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")


# ── cmd_update ─────────────────────────────────────────────────────────────

def cmd_update() -> None:
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    run_id    = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    log_path  = LOG_DIR / f"docker-update-{run_id}.log"
    json_path = LOG_DIR / f"docker-update-{run_id}.json"

    projects = discover_projects()
    if not projects:
        print("No Docker Compose projects found.")
        return

    results = _run_update_rich(projects) if HAS_RICH else _run_update_plain(projects)

    groups = _group_by_outcome(results)
    _print_summary_rich(groups) if HAS_RICH else _print_summary_plain(groups)

    _write_text_log(run_id, results, log_path)
    _write_json_log(run_id, results, json_path)

    print(f"Log saved to: {log_path}")
    print()
    print("To browse logs:  docker-update log")


# ── cmd_log ────────────────────────────────────────────────────────────────

def cmd_log() -> None:
    if not LOG_DIR.exists() or not any(LOG_DIR.glob("*.log")):
        print(f"No log files found in {LOG_DIR}")
        return

    fzf_bin    = shutil.which("fzf")
    helper_bin = shutil.which(FZF_HELPER)

    missing: list[str] = []
    if not fzf_bin:
        missing.append("fzf")
    if not helper_bin:
        missing.append(FZF_HELPER)

    if missing:
        print(f"Warning: {', '.join(missing)} not found — falling back to plain-text log viewer.")
        print()
        _cmd_log_plain()
        return

    os.execvp(helper_bin, [helper_bin, "log", str(LOG_DIR)])


def _cmd_log_plain() -> None:
    logs = sorted(LOG_DIR.glob("*.log"), reverse=True)
    if not logs:
        print("No logs found.")
        return

    print("Available logs (newest first):")
    print()
    for i, log in enumerate(logs, 1):
        print(f"  {i:3d}.  {log.name}")

    print()
    try:
        raw = input("Enter number to view (or Enter to exit): ").strip()
    except (EOFError, KeyboardInterrupt):
        print()
        return

    if not raw:
        return

    try:
        selected = logs[int(raw) - 1]
        print()
        print("─" * 60)
        print(selected.read_text(encoding="utf-8"))
    except (ValueError, IndexError):
        print("Invalid selection.")


# ── cmd_debug ──────────────────────────────────────────────────────────────

def cmd_debug() -> None:
    """Dry-run: discover projects and show what would happen, no changes."""
    projects = discover_projects()
    if not projects:
        print("No Docker Compose projects found.")
        return

    if HAS_RICH:
        _debug_rich(projects)
    else:
        _debug_plain(projects)

    print()
    try:
        answer = input("Show example log in fzf? [y/N] ").strip().lower()
    except (EOFError, KeyboardInterrupt):
        print()
        return

    if answer != "y":
        return

    fzf_bin    = shutil.which("fzf")
    helper_bin = shutil.which(FZF_HELPER)

    missing = []
    if not fzf_bin:
        missing.append("fzf")
    if not helper_bin:
        missing.append(FZF_HELPER)

    if missing:
        print(f"Cannot open fzf demo: {', '.join(missing)} not found.")
        return

    import random
    import tempfile

    # Assign fake outcomes weighted towards success
    outcome_pool = (
        [SUCCEEDED] * 6
        + [FAILED_PULL] * 2
        + [FAILED_STOP]
        + [FAILED_RESTART]
    )
    fake_projects = [
        {
            "name":    p["name"],
            "workdir": p["workdir"],
            "is_git":  p["is_git"],
            "outcome": random.choice(outcome_pool),
        }
        for p in projects
    ]

    fake_data = {
        "run_id":  "debug-" + datetime.now().strftime("%Y-%m-%d_%H-%M-%S"),
        "debug":   True,
        "projects": fake_projects,
    }

    tmp = tempfile.NamedTemporaryFile(
        mode="w", suffix=".json", prefix="du-debug-", delete=False
    )
    json.dump(fake_data, tmp, indent=2)
    tmp.flush()
    tmp.close()

    os.execvp(helper_bin, [helper_bin, "debug", tmp.name])


def _debug_rich(projects: list[dict[str, Any]]) -> None:
    console.print()
    console.rule("[bold cyan]Debug Mode — no changes will be made")
    console.print()

    tbl = Table(show_header=True, header_style="bold")
    tbl.add_column("Project",     style="cyan",    no_wrap=True)
    tbl.add_column("Workdir",     style="dim")
    tbl.add_column("Type")
    tbl.add_column("Update steps")

    for p in projects:
        ptype  = "[blue]git[/blue]" if p["is_git"] else "[magenta]registry[/magenta]"
        steps  = "git pull → up -d --build" if p["is_git"] else "down → pull → up -d"
        tbl.add_row(p["name"], p["workdir"], ptype, steps)

    console.print(tbl)
    console.print()
    console.print(f"  [dim]{len(projects)} project(s) would be updated[/dim]")


def _debug_plain(projects: list[dict[str, Any]]) -> None:
    print()
    print("Debug Mode — no changes will be made")
    print("=" * 60)
    for p in projects:
        ptype = "git" if p["is_git"] else "registry"
        steps = "git pull → up -d --build" if p["is_git"] else "down → pull → up -d"
        print(f"  {p['name']}")
        print(f"    workdir : {p['workdir']}")
        print(f"    type    : {ptype}")
        print(f"    steps   : {steps}")
        print()
    print(f"{len(projects)} project(s) would be updated")


# ── entrypoint ─────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(
        prog="docker-update",
        description="Docker Compose stack manager",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            commands:
              (none)   update all stacks
              log      browse past update logs
              debug    dry-run — show what would run without making changes
        """),
    )
    parser.add_argument(
        "command",
        nargs="?",
        choices=["log", "debug"],
        default=None,
    )
    args = parser.parse_args()

    if args.command == "log":
        cmd_log()
    elif args.command == "debug":
        cmd_debug()
    else:
        cmd_update()


if __name__ == "__main__":
    main()