Projektwissen der AG Prozesse aus AG_QGIS/Projektwissen abgeleitet
Ü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>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,626 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Lesender Zugriff auf das Gitea-Ticketsystem der Organisation aus workspace.toml.
|
||||
|
||||
Fragt die öffentlich lesbare Gitea-API ab (kein Token, keine Anmeldung, keine
|
||||
Installation – nur Standardbibliothek) und gibt Issues, Pull Requests und
|
||||
Suchtreffer token-sparend für KI-Agenten und Entwickler aus. Läuft auf Linux,
|
||||
macOS und Windows mit jedem Python >= 3.11.
|
||||
|
||||
WICHTIGE PROJEKTREGEL: Agenten dürfen auf dem Gitea-Server nichts verändern.
|
||||
Dieses Skript stellt ausschließlich GET-Anfragen. Schreibende Aktionen (neues
|
||||
Issue, neuer Kommentar) werden nur als Entwurfsdatei unter entwuerfe/ vorbereitet.
|
||||
Ein Mensch prüft den Text und sendet ihn selbst über die Weboberfläche ab.
|
||||
|
||||
Aufrufe (aus dem Workspace-Root):
|
||||
python scripts/gitea.py tickets # offene Issues ALLER Repos
|
||||
python scripts/gitea.py tickets Projektwissen # nur ein Repo (Kurzname genügt)
|
||||
python scripts/gitea.py tickets --geschlossen # geschlossene statt offene
|
||||
python scripts/gitea.py tickets --label Bug # nach Label filtern
|
||||
python scripts/gitea.py ticket Projektwissen 12 # ein Issue mit allen Kommentaren
|
||||
python scripts/gitea.py suche "Swimlane" # Volltext über Titel, Text und Kommentare,
|
||||
# offene und geschlossene Tickets
|
||||
python scripts/gitea.py pulls # offene Pull Requests
|
||||
python scripts/gitea.py entwurf issue Projektwissen # Entwurfsdatei anlegen
|
||||
python scripts/gitea.py entwurf kommentar Projektwissen 12 # Entwurf für einen Kommentar
|
||||
|
||||
Exit-Code: 0 = in Ordnung, 1 = Gitea nicht erreichbar / nicht gefunden,
|
||||
2 = falsche Angabe (Repo unbekannt oder mehrdeutig).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import shutil
|
||||
import tomllib
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MANIFEST = ROOT / "workspace.toml"
|
||||
ENTWURFSORDNER = ROOT / "entwuerfe"
|
||||
VORLAGE_ISSUE = ROOT / "docs" / "vorlagen" / "issue.md"
|
||||
|
||||
BENUTZERKENNUNG = "AG_Prozesse-Projektwissen/1.0"
|
||||
ZEITLIMIT = 20 # Sekunden je Anfrage
|
||||
SEITENGROESSE = 50 # Gitea liefert sonst nur wenige Einträge je Seite
|
||||
MAX_SEITEN = 20 # Sicherheitsgrenze gegen Endlosschleifen
|
||||
LEER = "-"
|
||||
|
||||
|
||||
class GiteaFehler(Exception):
|
||||
"""Netz-, HTTP- oder Formatfehler beim Zugriff auf die API (Exit-Code 1)."""
|
||||
|
||||
|
||||
class NichtGefunden(GiteaFehler):
|
||||
"""HTTP 404: Repo oder Issue existiert nicht bzw. hat kein Ticketsystem."""
|
||||
|
||||
|
||||
class EingabeFehler(Exception):
|
||||
"""Unbekannter oder mehrdeutiger Repo-Name (Exit-Code 2)."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Manifest ---
|
||||
|
||||
def lade_manifest() -> dict:
|
||||
with MANIFEST.open("rb") as f:
|
||||
return tomllib.load(f)
|
||||
|
||||
|
||||
def zugangsdaten(manifest: dict) -> tuple[str, str]:
|
||||
"""Basis-URL und Organisation aus [workspace] – nie hart kodieren."""
|
||||
ws = manifest.get("workspace", {})
|
||||
basis = str(ws.get("gitea", "")).rstrip("/")
|
||||
org = str(ws.get("org", ""))
|
||||
if not basis or not org:
|
||||
raise GiteaFehler("In workspace.toml fehlen [workspace] gitea oder org.")
|
||||
return basis, org
|
||||
|
||||
|
||||
def repo_namen(manifest: dict) -> list[str]:
|
||||
"""Das Workspace-Repo ([workspace] repo) zuerst, danach die Klone aus [[repos]]."""
|
||||
eigenes = manifest.get("workspace", {}).get("repo")
|
||||
namen = [eigenes] if eigenes else []
|
||||
return namen + [r["name"] for r in manifest.get("repos", []) if r.get("name")]
|
||||
|
||||
|
||||
def loese_repo(kurzname: str, namen: list[str]) -> str:
|
||||
"""Löst einen Kurznamen wie 'wissen' zu 'Projektwissen' auf.
|
||||
|
||||
Reihenfolge: exakter Name gewinnt, dann Groß-/Kleinschreibung egal, dann
|
||||
eindeutige Übereinstimmung am Namensende, dann eindeutiger Teilstring.
|
||||
"""
|
||||
if kurzname in namen:
|
||||
return kurzname
|
||||
klein = kurzname.lower()
|
||||
for pruefung in (
|
||||
lambda n: n.lower() == klein,
|
||||
lambda n: n.lower().endswith(klein),
|
||||
lambda n: klein in n.lower(),
|
||||
):
|
||||
treffer = [n for n in namen if pruefung(n)]
|
||||
if len(treffer) == 1:
|
||||
return treffer[0]
|
||||
if len(treffer) > 1:
|
||||
raise EingabeFehler(
|
||||
f"Repo-Name '{kurzname}' ist mehrdeutig. Gemeint sein könnte: "
|
||||
+ ", ".join(treffer)
|
||||
)
|
||||
raise EingabeFehler(
|
||||
f"Repo '{kurzname}' steht nicht in workspace.toml. Bekannt: " + ", ".join(namen)
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- API ------
|
||||
|
||||
def baue_url(basis: str, pfad: str, parameter: dict[str, str] | None = None) -> str:
|
||||
url = f"{basis}/api/v1{pfad}"
|
||||
if parameter:
|
||||
url += "?" + urllib.parse.urlencode(parameter)
|
||||
return url
|
||||
|
||||
|
||||
def hole_json(url: str) -> object:
|
||||
"""Eine einzelne GET-Anfrage. Schreibende Methoden gibt es hier bewusst nicht."""
|
||||
anfrage = urllib.request.Request(
|
||||
url, method="GET",
|
||||
headers={"User-Agent": BENUTZERKENNUNG, "Accept": "application/json"},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(anfrage, timeout=ZEITLIMIT) as antwort:
|
||||
roh = antwort.read()
|
||||
except urllib.error.HTTPError as fehler:
|
||||
if fehler.code == 404:
|
||||
raise NichtGefunden("Repo/Issue nicht gefunden: " + url) from None
|
||||
raise GiteaFehler(
|
||||
f"Gitea antwortet mit HTTP {fehler.code} ({fehler.reason}) auf {url}"
|
||||
) from None
|
||||
except TimeoutError:
|
||||
raise GiteaFehler(
|
||||
f"Gitea nicht erreichbar: Zeitlimit von {ZEITLIMIT} s überschritten. "
|
||||
"Netzverbindung oder Proxy prüfen."
|
||||
) from None
|
||||
except urllib.error.URLError as fehler:
|
||||
raise GiteaFehler(
|
||||
f"Gitea nicht erreichbar: {fehler.reason}. Netzverbindung oder Proxy prüfen."
|
||||
) from None
|
||||
except OSError as fehler:
|
||||
raise GiteaFehler(
|
||||
f"Gitea nicht erreichbar: {fehler}. Netzverbindung oder Proxy prüfen."
|
||||
) from None
|
||||
try:
|
||||
return json.loads(roh.decode("utf-8"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||||
raise GiteaFehler(f"Unerwartete Antwort (kein JSON) von {url}") from None
|
||||
|
||||
|
||||
def hole_liste(basis: str, pfad: str, parameter: dict[str, str]) -> list[dict]:
|
||||
"""Blättert über die Seiten, bis eine leere oder unvollständige Seite kommt."""
|
||||
ergebnisse: list[dict] = []
|
||||
for seite in range(1, MAX_SEITEN + 1):
|
||||
abfrage = dict(parameter, limit=str(SEITENGROESSE), page=str(seite))
|
||||
daten = hole_json(baue_url(basis, pfad, abfrage))
|
||||
if not isinstance(daten, list):
|
||||
# Gitea liefert bei deaktiviertem Ticketsystem ein Objekt statt einer Liste.
|
||||
raise NichtGefunden("Repo/Issue nicht gefunden: " + baue_url(basis, pfad, abfrage))
|
||||
ergebnisse.extend(daten)
|
||||
if len(daten) < SEITENGROESSE:
|
||||
break
|
||||
else:
|
||||
print(f"Hinweis: Sicherheitsgrenze von {MAX_SEITEN} Seiten erreicht – Liste evtl. unvollständig.")
|
||||
return ergebnisse
|
||||
|
||||
|
||||
def hole_issues(basis: str, org: str, repo: str, state: str, art: str = "issues") -> list[dict]:
|
||||
"""Issues EINES Repos. art='issues' blendet Pull Requests aus, 'pulls' zeigt nur sie."""
|
||||
pfad = f"/repos/{urllib.parse.quote(org)}/{urllib.parse.quote(repo)}/issues"
|
||||
try:
|
||||
return hole_liste(basis, pfad, {"state": state, "type": art})
|
||||
except NichtGefunden:
|
||||
raise NichtGefunden(
|
||||
f"Repo/Issue nicht gefunden: {org}/{repo} – Repo existiert nicht oder "
|
||||
"hat auf Gitea kein Ticketsystem."
|
||||
) from None
|
||||
|
||||
|
||||
def hole_issues_org(basis: str, org: str, state: str, art: str = "issues",
|
||||
suchtext: str | None = None) -> list[dict]:
|
||||
"""Issues ALLER Repos der Organisation über den Sammelendpunkt.
|
||||
|
||||
Spart pro Aufruf zehn Einzelabfragen; Repos ohne Ticketsystem fallen dabei
|
||||
automatisch weg. Mit suchtext sucht Gitea serverseitig (Titel, Text und
|
||||
Kommentare).
|
||||
"""
|
||||
parameter = {"owner": org, "state": state, "type": art}
|
||||
if suchtext:
|
||||
parameter["q"] = suchtext
|
||||
return hole_liste(basis, "/repos/issues/search", parameter)
|
||||
|
||||
|
||||
def sammle_issues(basis: str, org: str, repo: str | None, namen: list[str], state: str,
|
||||
art: str = "issues", suchtext: str | None = None) -> list[tuple[str, list[dict]]]:
|
||||
"""Holt die Einträge und gruppiert sie in der Reihenfolge aus workspace.toml."""
|
||||
if suchtext:
|
||||
# Auch für ein einzelnes Repo über den Sammelendpunkt: nur er durchsucht
|
||||
# zusätzlich die Kommentare.
|
||||
eintraege = hole_issues_org(basis, org, state, art, suchtext)
|
||||
if repo:
|
||||
eintraege = [e for e in eintraege
|
||||
if (e.get("repository") or {}).get("name") == repo]
|
||||
elif repo:
|
||||
eintraege = hole_issues(basis, org, repo, state, art)
|
||||
else:
|
||||
eintraege = hole_issues_org(basis, org, state, art)
|
||||
return gruppiere_nach_repo(eintraege, namen)
|
||||
|
||||
|
||||
def gruppiere_nach_repo(eintraege: list[dict], namen: list[str]) -> list[tuple[str, list[dict]]]:
|
||||
nach_repo: dict[str, list[dict]] = {}
|
||||
for eintrag in eintraege:
|
||||
name = (eintrag.get("repository") or {}).get("name") or "(unbekanntes Repo)"
|
||||
nach_repo.setdefault(name, []).append(eintrag)
|
||||
reihenfolge = [n for n in namen if n in nach_repo]
|
||||
reihenfolge += sorted(n for n in nach_repo if n not in namen)
|
||||
return [(n, nach_repo[n]) for n in reihenfolge]
|
||||
|
||||
|
||||
def enthaelt(eintrag: dict, suchtext: str) -> bool:
|
||||
"""Lokaler Volltextvergleich über Titel und Text (ohne Kommentare)."""
|
||||
klein = suchtext.lower()
|
||||
return klein in (eintrag.get("title") or "").lower() or klein in (eintrag.get("body") or "").lower()
|
||||
|
||||
|
||||
def hole_kommentare(basis: str, org: str, repo: str, nummer: int) -> list[dict]:
|
||||
pfad = f"/repos/{urllib.parse.quote(org)}/{urllib.parse.quote(repo)}/issues/{nummer}/comments"
|
||||
return hole_liste(basis, pfad, {})
|
||||
|
||||
|
||||
def hole_issue(basis: str, org: str, repo: str, nummer: int) -> dict:
|
||||
pfad = f"/repos/{urllib.parse.quote(org)}/{urllib.parse.quote(repo)}/issues/{nummer}"
|
||||
try:
|
||||
daten = hole_json(baue_url(basis, pfad))
|
||||
except NichtGefunden:
|
||||
raise NichtGefunden(f"Repo/Issue nicht gefunden: {repo} #{nummer}") from None
|
||||
if not isinstance(daten, dict) or "number" not in daten:
|
||||
raise NichtGefunden(f"Repo/Issue nicht gefunden: {repo} #{nummer}")
|
||||
return daten
|
||||
|
||||
|
||||
# ------------------------------------------------------------- Formatierung ---
|
||||
|
||||
def terminalbreite(vorgabe: int | None = None) -> int:
|
||||
breite = vorgabe or shutil.get_terminal_size((100, 24)).columns
|
||||
return max(60, min(breite, 200))
|
||||
|
||||
|
||||
def kuerze(text: str, breite: int) -> str:
|
||||
text = " ".join((text or "").split())
|
||||
if len(text) <= breite:
|
||||
return text
|
||||
return text[: max(1, breite - 1)] + "…"
|
||||
|
||||
|
||||
def zeitpunkt(wert: str | None) -> datetime | None:
|
||||
if not wert:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(wert.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def datum(wert: str | None) -> str:
|
||||
zeit = zeitpunkt(wert)
|
||||
return zeit.strftime("%Y-%m-%d") if zeit else LEER
|
||||
|
||||
|
||||
def datum_zeit(wert: str | None) -> str:
|
||||
zeit = zeitpunkt(wert)
|
||||
return zeit.strftime("%Y-%m-%d %H:%M") if zeit else LEER
|
||||
|
||||
|
||||
def alter_tage(wert: str | None) -> int:
|
||||
zeit = zeitpunkt(wert)
|
||||
if not zeit:
|
||||
return 0
|
||||
jetzt = datetime.now(timezone.utc)
|
||||
return max(0, (jetzt - zeit.astimezone(timezone.utc)).days)
|
||||
|
||||
|
||||
def labelnamen(eintrag: dict) -> list[str]:
|
||||
return [l.get("name", "") for l in (eintrag.get("labels") or [])]
|
||||
|
||||
|
||||
def zuweisung(eintrag: dict) -> list[str]:
|
||||
namen = [a.get("login", "") for a in (eintrag.get("assignees") or [])]
|
||||
if not namen and eintrag.get("assignee"):
|
||||
namen = [eintrag["assignee"].get("login", "")]
|
||||
return [n for n in namen if n]
|
||||
|
||||
|
||||
def mehrzahl(anzahl: int, einzahl: str, mehrzahlform: str) -> str:
|
||||
return f"{anzahl} {einzahl if anzahl == 1 else mehrzahlform}"
|
||||
|
||||
|
||||
def anzahl_tickets(anzahl: int, geschlossen: bool, pull: bool = False) -> str:
|
||||
"""Zählbare Wortform für die Summenzeile, z.B. '1 offenes Ticket'."""
|
||||
if pull:
|
||||
wort = ("offener Pull Request", "offene Pull Requests")
|
||||
if geschlossen:
|
||||
wort = ("geschlossener Pull Request", "geschlossene Pull Requests")
|
||||
elif geschlossen:
|
||||
wort = ("geschlossenes Ticket", "geschlossene Tickets")
|
||||
else:
|
||||
wort = ("offenes Ticket", "offene Tickets")
|
||||
return mehrzahl(anzahl, *wort)
|
||||
|
||||
|
||||
def zustandskuerzel(eintrag: dict) -> str:
|
||||
return "[offen]" if eintrag.get("state") == "open" else "[geschl.]"
|
||||
|
||||
|
||||
def autor(eintrag: dict) -> str:
|
||||
nutzer = eintrag.get("user") or {}
|
||||
login = nutzer.get("login", "unbekannt")
|
||||
voll = (nutzer.get("full_name") or "").strip()
|
||||
return f"{voll} ({login})" if voll and voll != login else login
|
||||
|
||||
|
||||
def tabellenkopf(breiten: dict[str, int]) -> str:
|
||||
return (" " + "Nr".rjust(breiten["nr"])
|
||||
+ " " + "Titel".ljust(breiten["titel"])
|
||||
+ " " + "Labels".ljust(breiten["labels"])
|
||||
+ " " + "Zugewiesen".ljust(breiten["zug"])
|
||||
+ " " + "Alter".rjust(breiten["alter"]))
|
||||
|
||||
|
||||
def spaltenbreiten(breite: int, eintraege: list[dict]) -> dict[str, int]:
|
||||
nr = max(4, max((len(str(e.get("number", 0))) + 1 for e in eintraege), default=4))
|
||||
labels, zug, alter = 16, 11, 5
|
||||
titel = breite - (nr + labels + zug + alter + 10)
|
||||
return {"nr": nr, "titel": max(20, titel), "labels": labels, "zug": zug, "alter": alter}
|
||||
|
||||
|
||||
def tabellenzeile(eintrag: dict, breiten: dict[str, int], zusatz: str = "") -> str:
|
||||
zeile = (" " + ("#" + str(eintrag.get("number", 0))).rjust(breiten["nr"])
|
||||
+ " " + kuerze(eintrag.get("title", ""), breiten["titel"]).ljust(breiten["titel"])
|
||||
+ " " + kuerze(", ".join(labelnamen(eintrag)) or LEER, breiten["labels"]).ljust(breiten["labels"])
|
||||
+ " " + kuerze(", ".join(zuweisung(eintrag)) or LEER, breiten["zug"]).ljust(breiten["zug"])
|
||||
+ " " + f"{alter_tage(eintrag.get('created_at'))}T".rjust(breiten["alter"]))
|
||||
return zeile + ("\n" + " " * (breiten["nr"] + 4) + zusatz if zusatz else "")
|
||||
|
||||
|
||||
def drucke_gruppen(gruppen: list[tuple[str, list[dict]]], breite: int, leermeldung: str,
|
||||
zusatz=None) -> tuple[int, int]:
|
||||
"""Gibt die nach Repo gruppierte Tabelle aus; Spaltenkopf nur einmal."""
|
||||
gruppen = [(repo, eintraege) for repo, eintraege in gruppen if eintraege]
|
||||
if not gruppen:
|
||||
print("\n" + leermeldung)
|
||||
return 0, 0
|
||||
breiten = spaltenbreiten(breite, [e for _, eintraege in gruppen for e in eintraege])
|
||||
print(tabellenkopf(breiten))
|
||||
gesamt = 0
|
||||
for repo, eintraege in gruppen:
|
||||
gesamt += len(eintraege)
|
||||
print(f"\n{repo} ({len(eintraege)})")
|
||||
for eintrag in sorted(eintraege, key=lambda e: -e.get("number", 0)):
|
||||
print(tabellenzeile(eintrag, breiten, zusatz(eintrag, breiten) if zusatz else ""))
|
||||
return gesamt, len(gruppen)
|
||||
|
||||
|
||||
def passt_zum_label(eintrag: dict, label: str | None) -> bool:
|
||||
if not label:
|
||||
return True
|
||||
return any(label.lower() == name.lower() for name in labelnamen(eintrag))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Befehle -----
|
||||
|
||||
def befehl_tickets(basis: str, org: str, repo: str | None, namen: list[str],
|
||||
args: argparse.Namespace) -> int:
|
||||
state = "closed" if args.geschlossen else "open"
|
||||
bezeichnung = "geschlossene" if args.geschlossen else "offene"
|
||||
gefunden = sammle_issues(basis, org, repo, namen, state)
|
||||
breite = terminalbreite(args.breite)
|
||||
|
||||
gruppen = [(name, [i for i in issues if passt_zum_label(i, args.label)])
|
||||
for name, issues in gefunden]
|
||||
filter_text = f", Label '{args.label}'" if args.label else ""
|
||||
print(f"{org} – {bezeichnung} Tickets{filter_text} (Stand {datetime.now():%Y-%m-%d})")
|
||||
gesamt, repos_mit = drucke_gruppen(gruppen, breite, "Keine passenden Tickets gefunden.")
|
||||
print(f"\nSumme: {anzahl_tickets(gesamt, args.geschlossen)} in {repos_mit} von "
|
||||
f"{mehrzahl(1 if repo else len(namen), 'Repo', 'Repos')} | Alter in Tagen"
|
||||
f" | Details: python scripts/gitea.py ticket <Repo> <Nr>")
|
||||
return 0
|
||||
|
||||
|
||||
def befehl_ticket(basis: str, org: str, repo: str, nummer: int) -> int:
|
||||
eintrag = hole_issue(basis, org, repo, nummer)
|
||||
kommentare = hole_kommentare(basis, org, repo, nummer)
|
||||
art = "Pull Request" if eintrag.get("pull_request") else "Issue"
|
||||
status = "offen" if eintrag.get("state") == "open" else "geschlossen"
|
||||
|
||||
print(f"# {repo} #{eintrag['number']} – {eintrag.get('title', '')}\n")
|
||||
print(f"- Art: {art}")
|
||||
print(f"- Status: {status}" + (f" (geschlossen am {datum(eintrag.get('closed_at'))})"
|
||||
if eintrag.get("closed_at") else ""))
|
||||
print(f"- Autor: {autor(eintrag)}")
|
||||
print(f"- Angelegt: {datum_zeit(eintrag.get('created_at'))}"
|
||||
f" (vor {alter_tage(eintrag.get('created_at'))} Tagen)"
|
||||
f", zuletzt geändert: {datum_zeit(eintrag.get('updated_at'))}")
|
||||
print(f"- Labels: {', '.join(labelnamen(eintrag)) or 'keine'}")
|
||||
print(f"- Zugewiesen: {', '.join(zuweisung(eintrag)) or 'niemandem'}")
|
||||
if eintrag.get("milestone"):
|
||||
print(f"- Meilenstein: {eintrag['milestone'].get('title', '')}")
|
||||
print(f"- URL: {eintrag.get('html_url', '')}\n")
|
||||
|
||||
print("## Beschreibung\n")
|
||||
print((eintrag.get("body") or "").strip() or "_(kein Text)_")
|
||||
|
||||
print(f"\n## Kommentare ({len(kommentare)})")
|
||||
if not kommentare:
|
||||
print("\n_(keine)_")
|
||||
for nr, kommentar in enumerate(kommentare, start=1):
|
||||
print(f"\n### {nr}. {autor(kommentar)} – {datum_zeit(kommentar.get('created_at'))}\n")
|
||||
print((kommentar.get("body") or "").strip() or "_(kein Text)_")
|
||||
return 0
|
||||
|
||||
|
||||
def fundstelle(eintrag: dict, begriff: str, breite: int) -> str:
|
||||
"""Kurzer Textausschnitt um die erste Fundstelle im Text."""
|
||||
text = " ".join((eintrag.get("body") or "").split())
|
||||
stelle = text.lower().find(begriff.lower())
|
||||
if stelle < 0:
|
||||
if begriff.lower() in (eintrag.get("title") or "").lower():
|
||||
return "(Treffer im Titel)"
|
||||
return "(Treffer in einem Kommentar)"
|
||||
start = max(0, stelle - breite // 3)
|
||||
ausschnitt = text[start : start + breite]
|
||||
return ("…" if start else "") + ausschnitt + ("…" if start + breite < len(text) else "")
|
||||
|
||||
|
||||
def befehl_suche(basis: str, org: str, repo: str | None, namen: list[str],
|
||||
args: argparse.Namespace) -> int:
|
||||
# Gesucht wird standardmäßig in offenen UND geschlossenen Tickets: die Antwort
|
||||
# auf eine Fachfrage steht oft in einem längst geschlossenen Ticket.
|
||||
state = "open" if args.offen else ("closed" if args.geschlossen else "all")
|
||||
gefunden = sammle_issues(basis, org, repo, namen, state, suchtext=args.begriff)
|
||||
if not any(issues for _, issues in gefunden):
|
||||
# Zweiter Versuch ohne Server-Suche: Gitea findet je nach Index nicht jeden Teilstring.
|
||||
gefunden = [(name, [i for i in issues if enthaelt(i, args.begriff)])
|
||||
for name, issues in sammle_issues(basis, org, repo, namen, state)]
|
||||
breite = terminalbreite(args.breite)
|
||||
|
||||
gruppen = [(name, [i for i in issues if passt_zum_label(i, args.label)])
|
||||
for name, issues in gefunden]
|
||||
|
||||
def ausschnitt(eintrag: dict, breiten: dict[str, int]) -> str:
|
||||
zustand = "" if state != "all" else zustandskuerzel(eintrag) + " "
|
||||
return zustand + fundstelle(eintrag, args.begriff, breite - breiten["nr"] - 10)
|
||||
|
||||
zustaende = {"open": "nur offene", "closed": "nur geschlossene", "all": "offene und geschlossene"}
|
||||
print(f"Suche nach '{args.begriff}' in {org} ({zustaende[state]} Tickets,"
|
||||
f" Titel, Text und Kommentare)")
|
||||
gesamt, _ = drucke_gruppen(gruppen, breite, "Keine Treffer.", ausschnitt)
|
||||
print(f"\nSumme: {gesamt} Treffer | Details: python scripts/gitea.py ticket <Repo> <Nr>")
|
||||
return 0
|
||||
|
||||
|
||||
def befehl_pulls(basis: str, org: str, repo: str | None, namen: list[str],
|
||||
args: argparse.Namespace) -> int:
|
||||
state = "closed" if args.geschlossen else "open"
|
||||
bezeichnung = "geschlossene" if args.geschlossen else "offene"
|
||||
gefunden = sammle_issues(basis, org, repo, namen, state, art="pulls")
|
||||
breite = terminalbreite(args.breite)
|
||||
|
||||
gruppen = [(name, [p for p in pulls if passt_zum_label(p, args.label)])
|
||||
for name, pulls in gefunden]
|
||||
print(f"{org} – {bezeichnung} Pull Requests (Stand {datetime.now():%Y-%m-%d})")
|
||||
gesamt, _ = drucke_gruppen(gruppen, breite, "Keine passenden Pull Requests gefunden.",
|
||||
lambda eintrag, breiten: f"von {autor(eintrag)}")
|
||||
print(f"\nSumme: {anzahl_tickets(gesamt, args.geschlossen, pull=True)} in "
|
||||
f"{mehrzahl(1 if repo else len(namen), 'Repo', 'Repos')}")
|
||||
return 0
|
||||
|
||||
|
||||
def befehl_entwurf(basis: str, org: str, repo: str, art: str, nummer: int | None) -> int:
|
||||
"""Legt eine Entwurfsdatei an. Abgeschickt wird sie ausschließlich von Hand."""
|
||||
zeitstempel = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
web = f"{basis}/{org}/{repo}"
|
||||
|
||||
if art == "issue":
|
||||
ziel = ENTWURFSORDNER / f"{zeitstempel}-issue-{repo}.md"
|
||||
kopf = f"# Entwurf: neues Issue in {org}/{repo}"
|
||||
vorlage = (VORLAGE_ISSUE.read_text(encoding="utf-8").strip()
|
||||
if VORLAGE_ISSUE.exists()
|
||||
else "## Ziel\n<Was soll am Ende möglich sein?>\n\n## Abnahmekriterien\n- [ ] <…>")
|
||||
quelle = ("docs/vorlagen/issue.md" if VORLAGE_ISSUE.exists()
|
||||
else "Notvorlage (docs/vorlagen/issue.md fehlt)")
|
||||
koerper = f"Titel: <Kurzer, sprechender Titel>\n\n{vorlage}"
|
||||
absenden = (f"1. Seite öffnen: {web}/issues/new\n"
|
||||
"2. Titelzeile und Text unterhalb von `--- ENTWURF ---` von Hand einfügen"
|
||||
" (die Zeile `Titel:` selbst nicht mitkopieren).\n"
|
||||
"3. Labels, Zuweisung und Meilenstein setzen und das Issue anlegen.")
|
||||
hinweis = f"Vorlage: {quelle}"
|
||||
else:
|
||||
eintrag = hole_issue(basis, org, repo, int(nummer))
|
||||
ziel = ENTWURFSORDNER / f"{zeitstempel}-kommentar-{repo}-{nummer}.md"
|
||||
kopf = f"# Entwurf: Kommentar zu {org}/{repo} #{nummer}"
|
||||
koerper = "<Kommentartext hier eintragen>"
|
||||
absenden = (f"1. Seite öffnen: {web}/issues/{nummer}\n"
|
||||
"2. Text unterhalb von `--- ENTWURF ---` in das Kommentarfeld einfügen.\n"
|
||||
"3. Kommentar abschicken.")
|
||||
hinweis = (f"Ticket: #{nummer} – {eintrag.get('title', '')}"
|
||||
f" (Status: {'offen' if eintrag.get('state') == 'open' else 'geschlossen'})")
|
||||
|
||||
inhalt = "\n".join([
|
||||
kopf,
|
||||
"",
|
||||
f"- Ziel-Repo: {org}/{repo} ({web})",
|
||||
f"- {hinweis}",
|
||||
f"- Angelegt: {datetime.now():%Y-%m-%d %H:%M} von scripts/gitea.py",
|
||||
"",
|
||||
"> **Dieser Entwurf wird NICHT automatisch abgeschickt.** KI-Agenten verändern auf",
|
||||
"> Gitea nichts. Ein Mensch prüft den Text und stellt ihn selbst über die",
|
||||
"> Weboberfläche ein.",
|
||||
"",
|
||||
"--- ENTWURF ---",
|
||||
"",
|
||||
koerper,
|
||||
"",
|
||||
"--- ENDE ENTWURF ---",
|
||||
"",
|
||||
"## Absenden (von Hand)",
|
||||
"",
|
||||
absenden,
|
||||
"",
|
||||
])
|
||||
|
||||
ENTWURFSORDNER.mkdir(parents=True, exist_ok=True)
|
||||
ziel.write_text(inhalt, encoding="utf-8")
|
||||
print(f"Entwurf angelegt: {ziel.relative_to(ROOT)}")
|
||||
print("Bitte prüfen, ergänzen und selbst auf Gitea einstellen – das Skript sendet nichts.")
|
||||
return 0
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ main ------
|
||||
|
||||
def main() -> int:
|
||||
breite_eltern = argparse.ArgumentParser(add_help=False)
|
||||
breite_eltern.add_argument("--breite", type=int, help="Terminalbreite erzwingen (Standard: automatisch)")
|
||||
|
||||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
unter = ap.add_subparsers(dest="befehl", required=True)
|
||||
|
||||
p = unter.add_parser("tickets", parents=[breite_eltern], help="Issues auflisten (Standard: alle Repos, offen)")
|
||||
p.add_argument("repo", nargs="?", help="Repo oder Kurzname (z.B. Projektwissen); ohne Angabe: alle")
|
||||
p.add_argument("--geschlossen", action="store_true", help="geschlossene statt offener Tickets")
|
||||
p.add_argument("--label", help="nur Tickets mit diesem Label (z.B. Bug)")
|
||||
|
||||
p = unter.add_parser("ticket", help="ein Issue mit allen Kommentaren als Markdown")
|
||||
p.add_argument("repo", help="Repo oder Kurzname (z.B. Projektwissen)")
|
||||
p.add_argument("nummer", type=int, help="Issue-Nummer")
|
||||
|
||||
p = unter.add_parser("suche", parents=[breite_eltern], help="Volltextsuche über Titel und Text")
|
||||
p.add_argument("begriff", help="Suchbegriff (Groß-/Kleinschreibung egal)")
|
||||
p.add_argument("repo", nargs="?", help="Repo oder Kurzname; ohne Angabe: alle")
|
||||
p.add_argument("--offen", action="store_true", help="nur offene Tickets (Standard: offene und geschlossene)")
|
||||
p.add_argument("--geschlossen", action="store_true", help="nur geschlossene Tickets")
|
||||
p.add_argument("--label", help="zusätzlich nach Label filtern")
|
||||
|
||||
p = unter.add_parser("pulls", parents=[breite_eltern], help="Pull Requests auflisten")
|
||||
p.add_argument("repo", nargs="?", help="Repo oder Kurzname; ohne Angabe: alle")
|
||||
p.add_argument("--geschlossen", action="store_true", help="geschlossene statt offener PRs")
|
||||
p.add_argument("--label", help="nur PRs mit diesem Label")
|
||||
|
||||
p = unter.add_parser("entwurf", help="Entwurfsdatei für einen Menschen vorbereiten (sendet nichts)")
|
||||
p.add_argument("art", choices=["issue", "kommentar"], help="Art des Entwurfs")
|
||||
p.add_argument("repo", help="Repo oder Kurzname (z.B. Projektwissen)")
|
||||
p.add_argument("nummer", nargs="?", type=int, help="Issue-Nummer (nur bei 'kommentar')")
|
||||
|
||||
args = ap.parse_args()
|
||||
|
||||
try:
|
||||
manifest = lade_manifest()
|
||||
basis, org = zugangsdaten(manifest)
|
||||
namen = repo_namen(manifest)
|
||||
if not namen:
|
||||
raise GiteaFehler("workspace.toml nennt weder [workspace] repo noch [[repos]]-Einträge.")
|
||||
|
||||
if args.befehl == "entwurf" and args.art == "kommentar" and args.nummer is None:
|
||||
print("Für 'entwurf kommentar' wird die Issue-Nummer gebraucht, "
|
||||
"z.B. python scripts/gitea.py entwurf kommentar Projektwissen 12")
|
||||
return 2
|
||||
|
||||
repo = loese_repo(args.repo, namen) if getattr(args, "repo", None) else None
|
||||
|
||||
if args.befehl == "tickets":
|
||||
return befehl_tickets(basis, org, repo, namen, args)
|
||||
if args.befehl == "ticket":
|
||||
return befehl_ticket(basis, org, repo, args.nummer)
|
||||
if args.befehl == "suche":
|
||||
return befehl_suche(basis, org, repo, namen, args)
|
||||
if args.befehl == "pulls":
|
||||
return befehl_pulls(basis, org, repo, namen, args)
|
||||
if args.befehl == "entwurf":
|
||||
return befehl_entwurf(basis, org, repo, args.art, args.nummer)
|
||||
except EingabeFehler as fehler:
|
||||
print(fehler)
|
||||
return 2
|
||||
except NichtGefunden as fehler:
|
||||
print(str(fehler))
|
||||
return 1
|
||||
except GiteaFehler as fehler:
|
||||
print(str(fehler))
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
print("Abgebrochen.")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,415 @@
|
||||
#!/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())
|
||||
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Findet deutschen Text in .py-Dateien, der offenbar ohne Umlaute/ß geschrieben wurde.
|
||||
|
||||
Hintergrund: AG_QGIS, Plugin_SN_Basis #42. KI-generierter Text weicht gelegentlich auf eine
|
||||
ASCII-Ersatzschreibweise aus ("gueltig" statt "gültig", "ueber" statt "über") statt
|
||||
echte Umlaute zu verwenden. Das betrifft vor allem Meldungstexte, Beschriftungen
|
||||
(etwa BPMN-Elementnamen) und Docstrings/Kommentare.
|
||||
|
||||
Technische Bezeichner (Variablen-, Funktions- und Schlüsselnamen) sind davon
|
||||
ausdrücklich nicht betroffen: Sie sind bewusst ASCII und enthalten kein Leerzeichen, während geschriebener deutscher Text so gut
|
||||
wie immer eines enthält. Deshalb prüft dieses Skript nur String-Inhalte mit
|
||||
mindestens einem Leerzeichen sowie Kommentartext - keine Bezeichner.
|
||||
|
||||
Kein Ersatz fürs Korrekturlesen: Die Liste verdächtiger Wortstämme ist bewusst kurz
|
||||
gehalten (nur eindeutige Fälle), um keine Fehlalarme zu erzeugen. Neue, hier nicht
|
||||
erfasste Wortstämme ergänzen.
|
||||
|
||||
Aufrufe (aus dem Workspace-Root):
|
||||
python scripts/pruefe_umlaute.py # alle .py-Dateien im Workspace
|
||||
python scripts/pruefe_umlaute.py .agents/skills # nur unterhalb dieser Ordner
|
||||
|
||||
Exit-Code: 0 = keine Funde, 1 = mindestens ein Fund.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
# Ordner, die nie geprüft werden (Git-Interna, gespiegelte Skills, lokale Entwürfe).
|
||||
AUSGESCHLOSSEN = {".git", ".claude", "__pycache__", "entwuerfe"}
|
||||
|
||||
# Dateien, die bewusst ASCII-Ersatzschreibweisen enthalten: hallo.py schreibt seine
|
||||
# Tabellenwerte absichtlich ohne Umlaute, dieses Skript nennt die Muster selbst.
|
||||
DATEIEN_OHNE_PRUEFUNG = {"scripts/hallo.py", "scripts/pruefe_umlaute.py"}
|
||||
|
||||
# Nur eindeutige Fälle: Wortstämme, die als ASCII-Ersatzschreibweise so gut wie nie
|
||||
# etwas anderes als ein fehlender Umlaut/ß sein können. Absichtlich kein "ss" für
|
||||
# "ß" (zu viele legitime Doppel-S-Wörter) und kein "ae/oe/ue" pauschal (zu viele
|
||||
# Fehlalarme in technischen Bezeichnern und Fremdwörtern).
|
||||
VERDAECHTIGE_STAEMME = [
|
||||
r"gueltig\w*", r"ungueltig\w*", r"verfuegbar\w*", r"ausgewaehlt\w*",
|
||||
r"gewaehlt\w*", r"waehlen\b", r"auswaehlen\b", r"loeschen\b", r"loescht\w*",
|
||||
r"groesse\w*", r"zuruecksetzen\b", r"erlaeuter\w*", r"bestaetig\w*",
|
||||
r"natuerlich\w*", r"moeglich\w*", r"zusaetzlich\w*", r"abhaengig\w*",
|
||||
r"unzulaessig\w*", r"ueberpruef\w*", r"vollstaendig\w*", r"geaendert\w*",
|
||||
r"aenderung\w*", r"einfuegen\b", r"ausfuehr\w*", r"benoetig\w*",
|
||||
r"gewuenscht\w*", r"ueber\w*", r"fuer\b", r"hoehe\w*", r"waere\b",
|
||||
]
|
||||
_STAMM_PATTERN = re.compile(
|
||||
r"\b(?:" + "|".join(VERDAECHTIGE_STAEMME) + r")\b", re.IGNORECASE
|
||||
)
|
||||
|
||||
# Erlaubt genau diese exakten Fundstellen (Datei, Zeile) - für Ausnahmefälle, die
|
||||
# geprüft und bewusst akzeptiert wurden. Aktuell leer: siehe Docstring, warum
|
||||
# technische Bezeichner die Heuristik ohnehin nicht erreichen.
|
||||
ALLOWLIST: set[tuple[str, int]] = set()
|
||||
|
||||
_STRING_PATTERN = re.compile(
|
||||
r'"""(?P<triple_d>(?:.|\n)*?)"""'
|
||||
r"|'''(?P<triple_s>(?:.|\n)*?)'''"
|
||||
r'|"(?P<double>(?:[^"\\\n]|\\.)*)"'
|
||||
r"|'(?P<single>(?:[^'\\\n]|\\.)*)'"
|
||||
)
|
||||
_KOMMENTAR_PATTERN = re.compile(r"#(?P<text>.*)$", re.MULTILINE)
|
||||
|
||||
|
||||
def _kandidaten(quelltext: str):
|
||||
"""Liefert (start_offset, text) für String-Inhalte und Kommentare mit Leerzeichen."""
|
||||
for m in _STRING_PATTERN.finditer(quelltext):
|
||||
text = next(g for g in m.groups() if g is not None)
|
||||
if " " in text:
|
||||
yield m.start(), text
|
||||
for m in _KOMMENTAR_PATTERN.finditer(quelltext):
|
||||
text = m.group("text")
|
||||
if " " in text:
|
||||
yield m.start("text"), text
|
||||
|
||||
|
||||
def pruefe_datei(pfad: Path) -> list[tuple[int, str, str]]:
|
||||
"""Liefert Funde als (Zeilennummer, Wortstamm, Kontext) für eine Datei.
|
||||
|
||||
Ein Kandidat (String-Inhalt oder Kommentar) kann mehrzeilig sein (Docstrings);
|
||||
deshalb wird für jeden Treffer einzeln die Zeile anhand seiner eigenen Position
|
||||
berechnet, nicht anhand des Kandidaten-Anfangs.
|
||||
"""
|
||||
quelltext = pfad.read_text(encoding="utf-8", errors="replace")
|
||||
funde = []
|
||||
for offset, text in _kandidaten(quelltext):
|
||||
for treffer in _STAMM_PATTERN.finditer(text):
|
||||
zeile = quelltext.count("\n", 0, offset + treffer.start()) + 1
|
||||
zeilen_text = text.splitlines()[treffer.string.count("\n", 0, treffer.start())] \
|
||||
if "\n" in text else text
|
||||
kontext = zeilen_text.strip()
|
||||
if len(kontext) > 90:
|
||||
kontext = kontext[:87] + "..."
|
||||
funde.append((zeile, treffer.group(0), kontext))
|
||||
return funde
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("ordner", nargs="*", help="Ordner relativ zum Workspace-Root (Standard: alles)")
|
||||
args = ap.parse_args()
|
||||
|
||||
wurzeln = [ROOT / o for o in args.ordner] if args.ordner else [ROOT]
|
||||
fehlend = [w for w in wurzeln if not w.is_dir()]
|
||||
if fehlend:
|
||||
print("Ordner nicht gefunden: " + ", ".join(str(w) for w in fehlend))
|
||||
return 2
|
||||
|
||||
gesamt_funde = 0
|
||||
dateien = 0
|
||||
for wurzel in wurzeln:
|
||||
for py_datei in sorted(wurzel.rglob("*.py")):
|
||||
if any(teil in AUSGESCHLOSSEN for teil in py_datei.relative_to(ROOT).parts):
|
||||
continue
|
||||
relativ = py_datei.relative_to(ROOT).as_posix()
|
||||
if relativ in DATEIEN_OHNE_PRUEFUNG:
|
||||
continue
|
||||
dateien += 1
|
||||
for zeile, stamm, kontext in pruefe_datei(py_datei):
|
||||
if (relativ, zeile) in ALLOWLIST:
|
||||
continue
|
||||
print(f"{relativ}:{zeile}: '{stamm}' in: {kontext}")
|
||||
gesamt_funde += 1
|
||||
|
||||
status = "GRÜN" if gesamt_funde == 0 else "ROT"
|
||||
print(f"\n{status}: {gesamt_funde} Fund/e in {dateien} Datei(en).")
|
||||
return 1 if gesamt_funde else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,124 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Richtet den AG_Prozesse-Workspace ein bzw. aktualisiert ihn.
|
||||
|
||||
1. Klont fehlende Repositories aus workspace.toml in Unterordner (Default-Branch).
|
||||
2. Holt für vorhandene Klone den Stand vom Server (git fetch) und zeigt Branch,
|
||||
Vorsprung/Rückstand und ungesicherte Änderungen an. Ein Pull passiert nur
|
||||
mit --pull und nur bei sauberem Arbeitsverzeichnis (fast-forward).
|
||||
3. Spiegelt die Skills aus .agents/skills/ nach .claude/skills/ (Claude Code liest
|
||||
nur dort; der Agent-Skills-Standard und andere Werkzeuge lesen .agents/skills/).
|
||||
|
||||
Aufrufe:
|
||||
python scripts/setup_workspace.py # klonen + Status + Skills
|
||||
python scripts/setup_workspace.py --pull # zusätzlich fast-forward pull
|
||||
python scripts/setup_workspace.py --nur-skills
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
LOKALE_ÜBERSCHREIBUNG = ROOT / "workspace.local.toml"
|
||||
|
||||
|
||||
def repos_root() -> Path:
|
||||
"""Ordner für die weiteren Repository-Klone, siehe workspace.local.toml (nicht versioniert)."""
|
||||
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
|
||||
|
||||
|
||||
def git(*args: str, cwd: Path) -> tuple[int, str]:
|
||||
p = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, errors="replace")
|
||||
return p.returncode, (p.stdout + p.stderr).strip()
|
||||
|
||||
|
||||
def sync_skills() -> None:
|
||||
quelle = ROOT / ".agents" / "skills"
|
||||
ziel = ROOT / ".claude" / "skills"
|
||||
if not quelle.is_dir():
|
||||
print("Keine Skills in .agents/skills/ gefunden.")
|
||||
return
|
||||
ziel.parent.mkdir(exist_ok=True)
|
||||
# Alten Spiegel entfernen (Symlink, Junction oder Kopie)
|
||||
if ziel.is_symlink() or (ziel.exists() and not ziel.is_dir()):
|
||||
ziel.unlink()
|
||||
elif ziel.is_dir():
|
||||
try:
|
||||
os.rmdir(ziel) # leere Junction/Ordner
|
||||
except OSError:
|
||||
shutil.rmtree(ziel)
|
||||
art = "symlink"
|
||||
try:
|
||||
os.symlink(quelle, ziel, target_is_directory=True)
|
||||
except (OSError, NotImplementedError):
|
||||
art = "kopie"
|
||||
if os.name == "nt":
|
||||
try:
|
||||
import _winapi # type: ignore[attr-defined]
|
||||
|
||||
_winapi.CreateJunction(str(quelle), str(ziel))
|
||||
art = "junction"
|
||||
except Exception:
|
||||
pass
|
||||
if art == "kopie":
|
||||
shutil.copytree(quelle, ziel, ignore=shutil.ignore_patterns("__pycache__"))
|
||||
anzahl = len([p for p in quelle.iterdir() if (p / "SKILL.md").is_file()])
|
||||
print(f"Skills: {anzahl} aus .agents/skills/ nach .claude/skills/ gespiegelt ({art}).")
|
||||
if art == "kopie":
|
||||
print(" Hinweis: Kopie statt Link – nach Änderungen an .agents/skills/ dieses Skript erneut ausführen.")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("--pull", action="store_true", help="saubere Klone per fast-forward aktualisieren")
|
||||
ap.add_argument("--nur-skills", action="store_true", help="nur Skills spiegeln, kein Git")
|
||||
args = ap.parse_args()
|
||||
|
||||
if not args.nur_skills:
|
||||
with (ROOT / "workspace.toml").open("rb") as f:
|
||||
manifest = tomllib.load(f)
|
||||
ws = manifest["workspace"]
|
||||
basis_url = f"{ws['gitea']}/{ws['org']}"
|
||||
rroot = repos_root()
|
||||
print(f"Workspace: {ROOT}\nGitea: {basis_url}\n")
|
||||
if rroot != ROOT:
|
||||
print(f"Repositories liegen laut workspace.local.toml unter: {rroot}\n")
|
||||
rroot.mkdir(parents=True, exist_ok=True)
|
||||
for repo in manifest.get("repos", []):
|
||||
name, branch = repo["name"], repo["branch"]
|
||||
pfad = rroot / name
|
||||
if not (pfad / ".git").is_dir():
|
||||
print(f"[{name}] klone Branch {branch} ...")
|
||||
rc, out = git("clone", "--branch", branch, f"{basis_url}/{name}.git", name, cwd=rroot)
|
||||
print(" " + ("ok" if rc == 0 else f"FEHLER: {out}"))
|
||||
continue
|
||||
git("fetch", "--quiet", "--prune", cwd=pfad)
|
||||
_, aktuell = git("rev-parse", "--abbrev-ref", "HEAD", cwd=pfad)
|
||||
_, dirty = git("status", "--porcelain", cwd=pfad)
|
||||
n_dirty = len([z for z in dirty.splitlines() if z.strip()])
|
||||
_, ab = git("rev-list", "--left-right", "--count", f"HEAD...origin/{aktuell}", cwd=pfad)
|
||||
vor, _, zurueck = ab.partition("\t") if "\t" in ab else ("?", "", "?")
|
||||
meldung = f"[{name}] Branch {aktuell} (+{vor}/-{zurueck.strip()} zu origin), {n_dirty} ungesicherte Änderungen"
|
||||
if aktuell != branch:
|
||||
meldung += f" <- Hinweis: Manifest erwartet {branch}"
|
||||
if args.pull and n_dirty == 0 and aktuell == branch:
|
||||
rc, out = git("pull", "--ff-only", "--quiet", cwd=pfad)
|
||||
meldung += " -> pull " + ("ok" if rc == 0 else f"FEHLER: {out.splitlines()[-1] if out else rc}")
|
||||
print(meldung)
|
||||
print()
|
||||
sync_skills()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user