management/scripts/gruppenpruefung.py

#!/usr/bin/env python3
"""gruppenpruefung.py — Verbund-Prüfungen über die Gruppe (Lab-CI).

Gehört zur Stillstandsprüfungs-Familie: Befunde färben die Pipeline
rot (das ist die Alarmanlage), fehlender Zugang bricht ab statt still
zu überspringen, und jede Prüfung bildet einen real passierten Fall ab
(Design 2026-08-11, Muster A/B; ADR-0012/0013):

  1. Gruppenliste ↔ docs/components/ — die Projektliste wird zur
     Laufzeit gelesen, nie im Code gepflegt (Retro-Lehre); ein neues
     Repo wird Befund statt Lücke. (F-008/F-009)
  2. Pointer-Präsenz — jede active/staged-Komponente trägt CLAUDE.md.
     (F-011: 4 von 5 fehlten, niemand prüfte)
  3. Meilenstein- und Prioritätspflicht über ALLE offenen
     Gruppen-Issues. (realer Fall: gitops#61, entstanden 2026-08-11,
     ohne Meilenstein)
  4. Issue-Drift management ↔ docs/issues/ — Titel, Zustand,
     Meilenstein, Priorität, Status-Label. (F-001/F-017-Klasse)
  5. Git-Hygiene seit 2026-08-07 — Autor- und Committer-Zeit 12:00:00
     UTC, kanonische Identität; ausgenommen sind maschinelle Absender
     (MASCHINEN, per Adresse — ADR-0009). (F-002/F-003)

Demo-Modus für den Altbestand (Muster-B-Nachweis, Feldtest F-002):
  --hygiene-lokal <klonverzeichnis> [--seit JJJJ-MM-TT]
prüft die volle Historie lokaler Klone offline; nur eigene
Identitäten zählen (die fünf aus F-003).

Usage: GITLAB_TOKEN=… python3 scripts/gruppenpruefung.py [repo-root]
       python3 scripts/gruppenpruefung.py --hygiene-lokal DIR [--seit DATUM]
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import urllib.parse
import urllib.request
from pathlib import Path

API = "https://git.lab/api/v4"
GRUPPE = "axion1337.chat"
GRENZE = "2026-08-07"
KANONISCH = ("Thore Cimbal", "cfx@riot.8shield.net")
# Maschinelle Absender: ihre Commits entstehen ohne Menschen am Gerät, eine
# Zuschreibung an eine Person wäre falsch (ADR-0009, dort für den Rotations-Bot
# entschieden; dieselbe Begründung trägt für jeden weiteren Automaten).
# Gematcht wird die ADRESSE, nicht der Name: "Administrator" ist als Name viel zu
# generisch, um damit Befunde stillzulegen.
MASCHINEN = {
    "turn-secret-rotation@axion1337.chat",  # TURN-Secret-Rotation (CronJob im Cluster)
    "admin@axion1337.chat",                 # Wiki.js Git-Storage (Anzeigename "Administrator")
    "wiki@axion1337.chat",                  # Wiki.js Git-Storage (Dienst-Identität)
}
EIGENE_MAILS = {"cfx@riot.8shield.net", "gamemaster@axion1337.de",
                "cfxqriot.8shield.net", "Scrublord@Mac.Bad"}
STATUS_LABEL = {"open": None, "next": "status:next",
                "in-progress": "status:doing", "waiting": "status:wartet"}

befunde: list[str] = []
hinweise: list[str] = []


def api(pfad: str, token: str):
    daten, seite = [], 1
    while True:
        trenner = "&" if "?" in pfad else "?"
        req = urllib.request.Request(
            f"{API}/{pfad}{trenner}per_page=100&page={seite}",
            headers={"PRIVATE-TOKEN": token})
        with urllib.request.urlopen(req) as antwort:
            batch = json.load(antwort)
        if not isinstance(batch, list):
            return batch
        daten += batch
        if len(batch) < 100:
            return daten
        seite += 1


def frontmatter(p: Path) -> dict:
    zeilen = p.read_text(encoding="utf-8").splitlines()
    if not zeilen or zeilen[0] != "---":
        return {}
    meta = {}
    for z in zeilen[1:]:
        if z == "---":
            break
        m = re.match(r"^(\w+):\s*(.*)$", z)
        if m:
            meta[m.group(1)] = m.group(2).strip().strip('"')
    return meta


def titel(p: Path) -> str:
    for z in p.read_text(encoding="utf-8").splitlines():
        if z.startswith("# "):
            return z[2:].strip()
    return ""


def zeit_ok(iso: str) -> bool:
    # GitLab liefert z. B. 2026-08-11T12:00:00.000+00:00
    m = re.match(r"^\d{4}-\d{2}-\d{2}T(\d{2}:\d{2}:\d{2})(?:\.\d+)?"
                 r"(Z|[+-]\d{2}:?\d{2})$", iso)
    if not m:
        return False
    zeit, offset = m.groups()
    return zeit == "12:00:00" and offset in ("Z", "+00:00", "+0000")


def hygiene_lokal(klondir: Path, seit: str) -> int:
    gesamt = 0
    for repo in sorted(d for d in klondir.iterdir() if (d / ".git").exists()):
        umgebung = dict(os.environ, TZ="UTC")  # Session-1-Lehre: TZ am git-Prozess
        raus = subprocess.run(
            ["git", "-C", str(repo), "log", "--all",
             f"--since={seit}", "--date=format-local:%H:%M:%S",
             "--format=%ae\t%ad\t%cd"],
            capture_output=True, text=True, env=umgebung).stdout
        zaehler = 0
        for zeile in raus.splitlines():
            mail, ad, cd = zeile.split("\t")
            if mail in EIGENE_MAILS and (ad != "12:00:00" or cd != "12:00:00"):
                zaehler += 1
        if zaehler:
            befunde.append(f"{repo.name}: {zaehler} Commits eigener "
                           f"Identitäten mit Echtzeit-Stempel (seit {seit})")
        gesamt += zaehler
    return gesamt


def main() -> int:
    if "--hygiene-lokal" in sys.argv:
        klondir = Path(sys.argv[sys.argv.index("--hygiene-lokal") + 1])
        seit = (sys.argv[sys.argv.index("--seit") + 1]
                if "--seit" in sys.argv else GRENZE)
        gesamt = hygiene_lokal(klondir, seit)
        for b in befunde:
            print(f"BEFUND {b}")
        print(f"gruppenpruefung --hygiene-lokal: {len(befunde)} Befunde, "
              f"{gesamt} Commits betroffen")
        return 1 if befunde else 0

    token = os.environ.get("GITLAB_TOKEN", "")
    if not token:
        sys.exit("GITLAB_TOKEN fehlt — ohne Lesezugriff kann nichts "
                 "geprüft werden; Abbruch statt stillem Skip.")
    root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd()

    # 1) Gruppenliste (Laufzeit) ↔ docs/components/
    live = {p["path"]: p for p in
            api(f"groups/{GRUPPE}/projects?include_subgroups=false", token)
            if not p.get("archived")}
    erklaert = {}
    for f in sorted((root / "docs/components").glob("*.md")):
        if f.name == "template.md":
            continue
        erklaert[frontmatter(f)["slug"]] = frontmatter(f)
    for slug in sorted(set(live) - set(erklaert)):
        befunde.append(f"Projekt {slug} existiert in der Gruppe, ist aber "
                       f"in docs/components/ nicht deklariert")
    for slug in sorted(set(erklaert) - set(live)):
        befunde.append(f"docs/components/{slug}.md deklariert ein Projekt, "
                       f"das die Gruppe nicht (mehr) führt")

    # 2) Pointer-Präsenz (active/staged, außer management selbst)
    for slug, meta in sorted(erklaert.items()):
        if meta.get("phase") not in ("active", "staged") or slug == "management":
            continue
        ref = live.get(slug, {}).get("default_branch", "main")
        pfad = urllib.parse.quote(f"{GRUPPE}/{slug}", safe="")
        try:
            api(f"projects/{pfad}/repository/files/CLAUDE.md?ref={ref}", token)
        except Exception:
            befunde.append(f"{slug}: keine CLAUDE.md-Pointer-Datei "
                           f"(F-011; Rollout-Issue vorhanden)")

    # 3) Meilenstein-/Prioritätspflicht gruppenweit
    for i in api(f"groups/{GRUPPE}/issues?state=opened", token):
        ref = i["references"]["full"]
        ms = (i.get("milestone") or {}).get("title", "")
        prios = [l for l in i["labels"] if l.startswith("priority:")]
        if not re.match(r"^M[1-5]\b", ms):
            befunde.append(f"{ref}: offenes Issue ohne Meilenstein "
                           f"M1–M5 ({ms or 'keiner'})")
        if len(prios) != 1:
            befunde.append(f"{ref}: {len(prios)} priority-Labels statt 1")

    # 4) Issue-Drift management ↔ docs/issues/
    gl = {i["iid"]: i for i in
          api(f"projects/{urllib.parse.quote(GRUPPE + '/management', safe='')}"
              f"/issues?state=opened", token)}
    dateien = {}
    for f in sorted((root / "docs/issues").glob("[0-9]*.md")):
        meta = frontmatter(f)
        if meta.get("gitlab_iid"):
            dateien[int(meta["gitlab_iid"])] = (f, meta)
        elif meta.get("status") not in ("done", "rejected"):
            hinweise.append(f"{f.name}: noch nicht gespiegelt "
                            f"(gitlab_iid fehlt — erwartet bis zum "
                            f"ersten Spiegel-Lauf)")
    for iid in sorted(set(gl) - set(dateien)):
        befunde.append(f"management#{iid} ist offen auf GitLab, hat aber "
                       f"keine kanonische Datei (zweites Backlog!)")
    for iid, (f, meta) in sorted(dateien.items()):
        if iid not in gl:
            if meta.get("status") not in ("done", "rejected"):
                befunde.append(f"{f.name}: offen im Repo, aber auf GitLab "
                               f"geschlossen/fehlend — nachziehen")
            continue
        i = gl[iid]
        if titel(f) != i["title"].strip():
            befunde.append(f"{f.name}: Titel weicht von GitLab ab")
        ms = (i.get("milestone") or {}).get("title", "")
        if not ms.startswith(meta.get("milestone", "")):
            befunde.append(f"{f.name}: Meilenstein {meta.get('milestone')} "
                           f"↔ GitLab {ms or 'keiner'}")
        if f"priority:{meta.get('priority')}" not in i["labels"]:
            befunde.append(f"{f.name}: Priorität {meta.get('priority')} "
                           f"fehlt auf GitLab")
        soll = STATUS_LABEL.get(meta.get("status"))
        ist = [l for l in i["labels"] if l.startswith("status:")]
        if (soll and soll not in ist) or (not soll and ist):
            befunde.append(f"{f.name}: Status {meta.get('status')} ↔ "
                           f"GitLab-Labels {ist or 'keine'}")

    # 5) Git-Hygiene seit der Regel-Grenze (API)
    for slug, meta in sorted(erklaert.items()):
        if meta.get("phase") == "external":
            continue
        pfad = urllib.parse.quote(f"{GRUPPE}/{slug}", safe="")
        for c in api(f"projects/{pfad}/repository/commits"
                     f"?since={GRENZE}T00:00:00Z&all=true", token):
            if (c.get("author_email") or "").lower() in MASCHINEN:
                continue
            wer = f"{c.get('author_name')} <{c.get('author_email')}>"
            if not zeit_ok(c["authored_date"]) or not zeit_ok(c["committed_date"]):
                befunde.append(f"{slug} {c['short_id']}: Echtzeit-Stempel "
                               f"({wer})")
            elif (c.get("author_name"), c.get("author_email")) != KANONISCH:
                befunde.append(f"{slug} {c['short_id']}: nicht-kanonische "
                               f"Identität {wer}")

    for b in befunde:
        print(f"BEFUND {b}")
    for h in hinweise:
        print(f"HINWEIS {h}")
    print(f"gruppenpruefung: {len(befunde)} Befunde, "
          f"{len(hinweise)} Hinweise")
    return 1 if befunde else 0


if __name__ == "__main__":
    sys.exit(main())