Projektwissen/scripts/hallo.py

#!/usr/bin/env python3
"""Sitzungsbeginn: zeigt den Stand aller Klone und aktualisiert die Kanal-Branches.

Für jedes Repository aus workspace.toml – und für den Workspace-Ordner selbst, sofern
er ein Git-Repository ist – holt das Skript den Stand vom Server (git fetch --prune),
ermittelt Branch, ungesicherte Änderungen und den Abstand zu origin und macht genau
dann einen fast-forward-Pull, wenn das gefahrlos ist:

  * das Arbeitsverzeichnis ist sauber (git status --porcelain leer),
  * der aktuelle Branch ist der Kanal-Branch aus workspace.toml,
  * es gibt einen Upstream,
  * wir sind hinter origin und haben keine eigenen Commits davor.

In allen anderen Fällen wird nur berichtet und nichts verändert. Am Ende stehen eine
Tabelle (eine Zeile je Repository) und, falls nötig, konkrete Handlungsempfehlungen.

Absolut verboten und deshalb nirgends im Code enthalten:
    git merge, git rebase, git stash, git reset, git checkout/switch auf einen anderen
    Branch, git push, git tag sowie jedes Verwerfen von Änderungen.
Ungesicherte oder unversionierte Dateien werden nie angefasst. Die einzige schreibende
Git-Operation des Skripts ist `git pull --ff-only` unter den vier Bedingungen oben.

Aufrufe (aus dem Workspace-Root):
    python scripts/hallo.py                       # fetch + fast-forward-Pull + Bericht
    python scripts/hallo.py --nur-pruefen         # fetch + Bericht, kein Pull
    python scripts/hallo.py --ohne-netz           # nur lokaler Stand, kein fetch
    python scripts/hallo.py --repo Plugin_SN_Basis --repo Repository

Der Kanal-Branch des Workspace-Repos selbst kann in workspace.toml unter
[workspace] branch = "..." gesetzt werden; ohne Eintrag gilt sein aktueller Branch.

Exit-Code: 0 = nichts Kritisches, 1 = ein Repo-Ordner fehlt oder ein Git-Befehl ist
unerwartet gescheitert, 2 = Aufruffehler (unbekanntes Repo, workspace.toml unlesbar).
"""

from __future__ import annotations

import argparse
import os
import subprocess
import textwrap
import tomllib
from dataclasses import dataclass, field
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
MANIFEST = ROOT / "workspace.toml"

BREITE = 100          # Zielbreite der Tabelle (lesbar im 100-Zeichen-Terminal)
FETCH_TIMEOUT = 60    # Sekunden je git fetch, damit ein hängender Server nicht blockiert

# Git ohne interaktive Rückfragen: sonst wartet ein fetch ewig auf eine Passworteingabe.
GIT_UMGEBUNG = dict(os.environ, GIT_TERMINAL_PROMPT="0", GCM_INTERACTIVE="never")


def git(*args: str, cwd: Path, timeout: int | None = None) -> tuple[int, str]:
    """Führt git aus und liefert (Rückgabecode, Ausgabe). Wirft nie eine Ausnahme."""
    try:
        p = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True,
                           errors="replace", env=GIT_UMGEBUNG, timeout=timeout)
    except FileNotFoundError:
        return 127, "git wurde nicht gefunden (steht git im PATH?)"
    except subprocess.TimeoutExpired:
        return 124, f"Zeitüberschreitung nach {timeout} s"
    return p.returncode, (p.stdout + p.stderr).strip()


def kurz(text: str, breite: int = 70) -> str:
    """Letzte, gekürzte Zeile einer Git-Ausgabe – für einzeilige Meldungen."""
    zeilen = [z for z in text.splitlines() if z.strip()]
    letzte = zeilen[-1].strip() if zeilen else ""
    return letzte if len(letzte) <= breite else letzte[: breite - 1] + "…"


def kuerze(text: str, breite: int) -> str:
    return text if len(text) <= breite else text[: max(1, breite - 1)] + "~"


def umbrechen(zeile: str) -> str:
    """Bricht Fließtext auf BREITE um; Befehlszeilen (vier Leerzeichen Einzug) bleiben ganz."""
    if zeile.startswith("    ") or len(zeile) <= BREITE:
        return zeile
    einzug = " " * (len(zeile) - len(zeile.lstrip()))
    return textwrap.fill(zeile.strip(), width=BREITE, initial_indent=einzug,
                         subsequent_indent=einzug)


def commits(anzahl: int) -> str:
    return f"{anzahl} Commit" if anzahl == 1 else f"{anzahl} Commits"


def aenderungen(anzahl: int) -> str:
    return f"{anzahl} ungesicherte Änderung" if anzahl == 1 else f"{anzahl} ungesicherte Änderungen"


@dataclass
class Befund:
    """Zustand eines Klons – wird erst gefüllt, dann als Tabellenzeile ausgegeben."""

    name: str
    kanal: str
    pfad: Path
    ist_wurzel: bool = False
    vorhanden: bool = True
    lesbar: bool = True
    branch: str = "-"
    geaendert: int = 0
    hinter: int = 0
    vor: int = 0
    upstream: bool = False
    remote: bool = True
    offline: bool = False
    aktualisiert: bool = False
    zustand: str = "-"
    aktion: str = "-"
    kritisch: bool = False
    hinweise: list[str] = field(default_factory=list)

    def befehl(self, rest: str) -> str:
        """Kopierbarer Git-Befehl, aufzurufen aus dem Workspace-Root."""
        return f"git {rest}" if self.ist_wurzel else f"git -C {self.name} {rest}"

    @property
    def abstand(self) -> str:
        return "-" if not (self.vorhanden and self.upstream) else f"{self.hinter}/{self.vor}"


def zustand_ermitteln(b: Befund) -> str:
    """Ein Klartextwort für die Spalte Zustand (ohne Emoji, auch für KI-Agenten lesbar)."""
    if b.geaendert:
        return f"geaendert ({b.geaendert})"
    if b.branch == "HEAD":
        return "detached HEAD"
    if b.branch != b.kanal:
        return "fremder branch"
    if not b.upstream:
        return "kein upstream"
    if b.offline:
        return "offline"
    return "sauber"


def einlesen(b: Befund, ohne_netz: bool) -> None:
    """Nur lesende Git-Befehle: fetch, rev-parse, status, rev-list."""
    if not b.pfad.is_dir():
        b.vorhanden, b.zustand, b.aktion, b.kritisch = False, "fehlt", "Ordner fehlt", True
        return
    if not (b.pfad / ".git").exists():
        b.vorhanden, b.zustand, b.aktion, b.kritisch = False, "kein git-repo", "Ordner ohne .git", True
        return

    _, remotes = git("remote", cwd=b.pfad)
    b.remote = bool(remotes.strip())
    if not b.remote:
        b.hinweise.append("kein Remote konfiguriert – kein Abgleich mit origin möglich.")
    elif not ohne_netz:
        rc, ausgabe = git("fetch", "--prune", "--quiet", cwd=b.pfad, timeout=FETCH_TIMEOUT)
        if rc != 0:
            b.offline = True
            b.hinweise.append(f"origin nicht erreichbar: {kurz(ausgabe)}")

    rc, branch = git("rev-parse", "--abbrev-ref", "HEAD", cwd=b.pfad)
    if rc != 0:
        b.lesbar, b.zustand, b.aktion, b.kritisch = False, "git-fehler", f"FEHLER: {kurz(branch, 30)}", True
        b.hinweise.append(f"git rev-parse fehlgeschlagen: {kurz(branch)}")
        return
    b.branch = branch
    if not b.kanal:                       # Workspace-Repo ohne Vorgabe in workspace.toml
        b.kanal = branch

    rc, status = git("status", "--porcelain", cwd=b.pfad)
    if rc != 0:
        b.lesbar, b.zustand, b.aktion, b.kritisch = False, "git-fehler", f"FEHLER: {kurz(status, 30)}", True
        b.hinweise.append(f"git status fehlgeschlagen: {kurz(status)}")
        return
    b.geaendert = len([z for z in status.splitlines() if z.strip()])

    rc, _ = git("rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}", cwd=b.pfad)
    b.upstream = rc == 0
    if b.upstream:
        rc, zaehl = git("rev-list", "--left-right", "--count", "HEAD...@{upstream}", cwd=b.pfad)
        teile = zaehl.split()
        if rc == 0 and len(teile) == 2 and all(t.isdigit() for t in teile):
            b.vor, b.hinter = int(teile[0]), int(teile[1])
        else:
            b.kritisch = True
            b.hinweise.append(f"Abstand zu origin nicht bestimmbar: {kurz(zaehl)}")
    b.zustand = zustand_ermitteln(b)


def abgleichen(b: Befund, nur_pruefen: bool, ohne_netz: bool) -> None:
    """Entscheidet über den fast-forward-Pull. Bei jedem Zweifel: nichts tun, nur melden."""
    if not (b.vorhanden and b.lesbar):
        return
    if b.geaendert:
        b.aktion = "uebersprungen: Aenderungen vorhanden"
        return
    if b.branch == "HEAD":
        b.aktion = "uebersprungen: detached HEAD"
        return
    if b.branch != b.kanal:
        b.aktion = "uebersprungen: fremder branch"
        return
    if not b.upstream:
        b.aktion = "uebersprungen: kein upstream"
        return
    if b.offline:                     # ohne erfolgreichen fetch ist der Abstand nur geraten
        b.aktion = "uebersprungen: offline"
        return
    if b.vor and b.hinter:
        b.aktion = "uebersprungen: divergiert"
        return
    if b.vor:
        eigene = "1 eigener Commit" if b.vor == 1 else f"{b.vor} eigene Commits"
        b.aktion = f"uebersprungen: {eigene}"
        return
    if not b.hinter:
        b.aktion = "bereits aktuell"
        return
    if nur_pruefen or ohne_netz:
        b.aktion = f"hinter origin ({commits(b.hinter)})"
        return

    rc, ausgabe = git("pull", "--ff-only", "--quiet", cwd=b.pfad)
    if rc == 0:
        b.aktion = f"aktualisiert ({commits(b.hinter)})"
        b.aktualisiert, b.hinter = True, 0
        b.zustand = zustand_ermitteln(b)
    else:
        b.aktion, b.kritisch = "FEHLER: pull fehlgeschlagen", True
        b.hinweise.append(f"git pull --ff-only fehlgeschlagen: {kurz(ausgabe)}")


def tabelle_ausgeben(befunde: list[Befund]) -> None:
    """Eine Zeile je Repository, insgesamt hoechstens BREITE Zeichen breit."""
    spalten = [
        ("Repo", 34, [b.name for b in befunde]),
        ("Branch", 14, [b.branch for b in befunde]),
        ("Zustand", 16, [b.zustand for b in befunde]),
        ("hinter/vor", 10, [b.abstand for b in befunde]),
    ]
    breiten = [min(hoechst, max([len(kopf)] + [len(w) for w in werte]))
               for kopf, hoechst, werte in spalten]
    platz = max(20, BREITE - (sum(breiten) + len(breiten)))
    b_aktion = min(platz, max([len("Aktion")] + [len(b.aktion) for b in befunde]))

    def zeile(felder: list[str]) -> str:
        gefuellt = [kuerze(w, br).ljust(br) for w, br in zip(felder[:-1], breiten)]
        return " ".join(gefuellt) + " " + kuerze(felder[-1], b_aktion)

    kopfzeile = zeile(["Repo", "Branch", "Zustand", "hinter/vor", "Aktion"])
    print(kopfzeile)
    print("-" * (sum(breiten) + len(breiten) + b_aktion))
    for b in befunde:
        print(zeile([b.name, b.branch, b.zustand, b.abstand, b.aktion]).rstrip())


def empfehlungen(befunde: list[Befund], nur_pruefen: bool, ohne_netz: bool) -> list[str]:
    """Konkrete, kopierbare Handlungsempfehlungen – nur für das, was wirklich ansteht."""
    zeilen: list[str] = []
    fehlend = [b.name for b in befunde if b.zustand == "fehlt"]
    if fehlend:
        zeilen.append(f"Repository-Ordner fehlen: {', '.join(fehlend)}. Klonen mit:")
        zeilen.append("    python scripts/setup_workspace.py")
    ohne_git = [b.name for b in befunde if b.zustand == "kein git-repo"]
    if ohne_git:
        zeilen.append(f"Ordner ohne .git (kein Klon): {', '.join(ohne_git)}. Von Hand prüfen und "
                      "wegräumen, danach: python scripts/setup_workspace.py")
    for b in befunde:
        if not b.vorhanden:
            continue
        block: list[str] = []
        if b.offline:
            block.append("origin nicht erreichbar – Netz/VPN prüfen. Die Angaben zu hinter/vor "
                         "stammen vom letzten erfolgreichen fetch.")
        if not b.remote:
            block.append("kein Remote konfiguriert – gehört der Ordner wirklich zum Workspace?")
        if b.geaendert:
            block.append(f"{aenderungen(b.geaendert)} – dieses Skript fasst sie nicht an. "
                         "Selbst sichten, committen oder verwerfen:")
            block.append(f"    {b.befehl('status')}")
        if b.branch == "HEAD":
            block.append("HEAD ist losgelöst (kein Branch). Vor der Arbeit auf einen Branch wechseln – "
                         "das macht das Skript bewusst nicht selbst.")
        elif b.branch != b.kanal:
            block.append(f"steht auf Branch '{b.branch}' statt auf dem Kanal-Branch '{b.kanal}'. "
                         f"Das ist richtig so, solange daran gearbeitet wird: Kanal-Branches werden nie "
                         f"direkt bearbeitet, die Rückführung läuft über einen Pull Request auf Gitea. "
                         f"Ist der Branch fertig, den PR stellen; '{b.kanal}' wird von diesem Skript "
                         f"nur aktualisiert, wenn er ausgecheckt ist.")
        if b.upstream and b.vor and b.hinter:
            block.append(f"divergiert: {commits(b.vor)} lokal, {commits(b.hinter)} auf origin. "
                         "Weder Merge noch Rebase noch Reset macht dieses Skript – bitte selbst entscheiden.")
        elif b.upstream and b.vor:
            block.append(f"{commits(b.vor)} lokal, nicht gepusht. Pushen ist Sache des Menschen "
                         "und passiert nur nach Absprache:")
            block.append(f"    {b.befehl(f'push origin {b.branch}')}")
        if not b.upstream and b.remote and b.branch != "HEAD":
            block.append(f"Branch '{b.branch}' hat keinen Upstream. Erst nach Absprache und nur vom "
                         "Menschen:")
            block.append(f"    {b.befehl(f'push -u origin {b.branch}')}")
        if b.hinter and not b.geaendert and b.branch == b.kanal and not b.vor:
            grund = ("fehlendem Netzkontakt" if b.offline else "--nur-pruefen" if nur_pruefen
                     else "--ohne-netz" if ohne_netz else "einem Hindernis")
            block.append(f"{commits(b.hinter)} hinter origin, wegen {grund} nicht geholt. Nachholen:")
            block.append(f"    python scripts/hallo.py --repo {b.name}")
        elif b.hinter and (b.geaendert or b.branch != b.kanal):
            block.append(f"Kanal-Branch liegt {commits(b.hinter)} hinter origin – erst nach dem "
                         f"Aufräumen oben aktualisierbar.")
        for meldung in b.hinweise:
            if meldung.startswith(("origin nicht erreichbar", "kein Remote")):
                continue                      # oben bereits in Klartext ausgegeben
            block.append(meldung)
        if block:
            zeilen.append(f"{b.name}:")
            zeilen.extend(f"  {z}" if not z.startswith("    ") else z for z in block)
    return zeilen


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--nur-pruefen", action="store_true", help="kein Pull, nur Bericht (fetch findet statt)")
    ap.add_argument("--ohne-netz", action="store_true", help="kein fetch und kein Pull, nur lokaler Zustand")
    ap.add_argument("--repo", action="append", metavar="NAME", help="nur dieses Repo (mehrfach angebbar)")
    args = ap.parse_args()

    try:
        with MANIFEST.open("rb") as f:
            manifest = tomllib.load(f)
    except (OSError, tomllib.TOMLDecodeError) as fehler:
        print(f"workspace.toml nicht lesbar: {fehler}")
        return 2

    repos = manifest.get("repos", [])
    bekannt = [ROOT.name] + [r["name"] for r in repos]
    if args.repo:
        unbekannt = [n for n in args.repo if n not in bekannt]
        if unbekannt:
            print(f"Unbekannte Repositories: {', '.join(unbekannt)}")
            print(f"Bekannt sind: {', '.join(bekannt)}")
            return 2

    def gewuenscht(name: str) -> bool:
        return args.repo is None or name in args.repo

    befunde: list[Befund] = []
    wurzel_ist_repo = (ROOT / ".git").exists()
    if wurzel_ist_repo and gewuenscht(ROOT.name):
        befunde.append(Befund(name=ROOT.name, kanal=manifest.get("workspace", {}).get("branch", ""),
                              pfad=ROOT, ist_wurzel=True))
    befunde += [Befund(name=r["name"], kanal=r["branch"], pfad=ROOT / r["name"])
                for r in repos if gewuenscht(r["name"])]

    modus = ("ohne Netz – kein fetch, kein Pull" if args.ohne_netz
             else "nur prüfen – fetch, kein Pull" if args.nur_pruefen
             else "fetch und fast-forward-Pull")
    print(f"Hallo. Workspace: {ROOT}")
    print(f"Modus: {modus} | Repositories: {len(befunde)}")
    if not wurzel_ist_repo and gewuenscht(ROOT.name):
        print(f"Hinweis: '{ROOT.name}' selbst ist noch kein Git-Repository – geprüft werden nur die Klone.")
    if not args.ohne_netz:
        print("Hole den Stand von origin (git fetch --prune) ...")
    print()

    for b in befunde:
        einlesen(b, args.ohne_netz)
        abgleichen(b, args.nur_pruefen, args.ohne_netz)

    tabelle_ausgeben(befunde)

    aktualisiert = sum(1 for b in befunde if b.aktualisiert)
    aktuell = sum(1 for b in befunde if b.aktion == "bereits aktuell")
    offen = len(befunde) - aktualisiert - aktuell
    wort = "Repository" if len(befunde) == 1 else "Repositories"
    print(f"\n{len(befunde)} {wort}: {aktualisiert} aktualisiert, {aktuell} bereits aktuell, "
          f"{offen} mit Befund.")
    if args.ohne_netz:
        print("Ohne Netz gelaufen: hinter/vor stammt vom letzten fetch und kann veraltet sein.")

    zu_tun = empfehlungen(befunde, args.nur_pruefen, args.ohne_netz)
    if zu_tun:
        print("\nZu tun")
        print("-" * 6)
        for zeile in zu_tun:
            print(umbrechen(zeile))

    return 1 if any(b.kritisch for b in befunde) else 0


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