DanielandClaude Opus 5.5
012509b6d1
Übernommen ohne Historie aus AG_QGIS/Projektwissen (main @ decd2a0), Skill sn-bpmn vom Branch BPMN-Skill @ 6efcac4 (ADR 0001). - Skripte hallo.py, gitea.py, setup_workspace.py auf die Organisation AG_Prozesse umgestellt; das Workspace-Repo zählt über [workspace] repo als Ticket-Repo, auch ohne weitere [[repos]]. - pruefe_umlaute.py prüft jetzt die .py-Dateien des Workspace selbst. - Branchmodell main plus Arbeitsbranches (ADR 0002). - AGENTS.md, README, Leitfäden, Skills, Vorlagen, Fallstricke und Glossar auf die Prozessmodellierung zugeschnitten; QGIS- und Plugin-Spezifisches entfernt. - Tests: python -m pytest -q .agents/skills -> 32 bestanden. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
416 lines
18 KiB
Python
416 lines
18 KiB
Python
#!/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())
|