#!/usr/bin/env python3 """Sitzungsbeginn: zeigt den Stand aller Klone und aktualisiert die Hauptbranches. 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 Hauptbranch 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 Projektwissen Der Hauptbranch 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" LOKALE_ÜBERSCHREIBUNG = ROOT / "workspace.local.toml" def repos_root() -> Path: """Ordner, unter dem die weiteren Repositories aus [[repos]] liegen. Standard: ROOT selbst (Repos als direkte Unterordner, wie in workspace.toml dokumentiert). Wer die Klone woanders liegen hat, kann das lokal und nicht versioniert in workspace.local.toml überschreiben: [workspace] repos_root = "D:/Klone/AG_Prozesse" """ if not LOKALE_ÜBERSCHREIBUNG.is_file(): return ROOT with LOKALE_ÜBERSCHREIBUNG.open("rb") as f: lokal = tomllib.load(f) pfad = lokal.get("workspace", {}).get("repos_root") return Path(pfad).expanduser() if pfad else ROOT 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 Hauptbranch '{b.kanal}'. " f"Das ist richtig so, solange daran gearbeitet wird: Der Hauptbranch wird 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"Hauptbranch 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", []) # Das Workspace-Repo heißt wie in [workspace] repo angegeben, sonst wie sein Ordner. wurzel_name = manifest.get("workspace", {}).get("repo") or ROOT.name bekannt = [wurzel_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(wurzel_name): befunde.append(Befund(name=wurzel_name, kanal=manifest.get("workspace", {}).get("branch", ""), pfad=ROOT, ist_wurzel=True)) rroot = repos_root() befunde += [Befund(name=r["name"], kanal=r["branch"], pfad=rroot / 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}") if rroot != ROOT: print(f"Repositories liegen laut workspace.local.toml unter: {rroot}") print(f"Modus: {modus} | Repositories: {len(befunde)}") if not wurzel_ist_repo and gewuenscht(wurzel_name): print(f"Hinweis: '{wurzel_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())