management/scripts/spiegel_issues.py

#!/usr/bin/env python3
"""spiegel_issues.py — Repo → GitLab, ein deterministischer Spiegel.

docs/issues/ ist kanonisch (ADR-0012); dieses Skript bespielt die
GitLab-Ansicht, damit Board, Meilensteine und Labels weiterarbeiten.
Gespiegelt werden **nur** Titel, Zustand, Meilenstein, Priorität,
Fälligkeit und Status-Label — nie Beschreibungen (der Migrations-
Fußtext und die Kommentare auf GitLab bleiben unangetastet), und nie
in Gegenrichtung.

Default ist **Dry-Run**: druckt jeden geplanten API-Aufruf und ändert
nichts. `--ausfuehren` schreibt wirklich — das stößt nur sorb an.
GitLab-seitig offene Issues ohne kanonische Datei werden gemeldet,
nie automatisch geschlossen.

Usage: python3 scripts/spiegel_issues.py [repo-root] [--ausfuehren]
       (Token: ~/.config/gitlab-lab/token oder $GITLAB_TOKEN)
"""
from __future__ import annotations

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

API = "https://git.lab/api/v4"
PROJEKT = urllib.parse.quote("axion1337.chat/management", safe="")
GRUPPE = "axion1337.chat"
STATUS_LABEL = {"open": None, "next": "status:next",
                "in-progress": "status:doing", "waiting": "status:wartet"}


def token_lesen() -> str:
    if os.environ.get("GITLAB_TOKEN"):
        return os.environ["GITLAB_TOKEN"]
    return (Path.home() / ".config/gitlab-lab/token").read_text(
        encoding="utf-8").strip()


def api(pfad: str, token: str, methode: str = "GET", daten: dict | None = None):
    body = json.dumps(daten).encode() if daten is not None else None
    req = urllib.request.Request(
        f"{API}/{pfad}", data=body, method=methode,
        headers={"PRIVATE-TOKEN": token, "Content-Type": "application/json"})
    with urllib.request.urlopen(req) as antwort:
        return json.load(antwort)


def alle(pfad: str, token: str):
    daten, seite = [], 1
    while True:
        trenner = "&" if "?" in pfad else "?"
        batch = api(f"{pfad}{trenner}per_page=100&page={seite}", token)
        daten += batch
        if len(batch) < 100:
            return daten
        seite += 1


def frontmatter_und_titel(p: Path):
    zeilen = p.read_text(encoding="utf-8").splitlines()
    meta, titel, i = {}, "", 1
    while i < len(zeilen) and zeilen[i] != "---":
        m = re.match(r"^(\w+):\s*(.*)$", zeilen[i])
        if m:
            meta[m.group(1)] = m.group(2).strip().strip('"')
        i += 1
    for z in zeilen[i:]:
        if z.startswith("# "):
            titel = z[2:].strip()
            break
    return meta, titel


def main() -> int:
    argv = [a for a in sys.argv[1:] if a != "--ausfuehren"]
    scharf = "--ausfuehren" in sys.argv
    root = Path(argv[0]) if argv else Path.cwd()
    token = token_lesen()

    meilensteine = {re.match(r"^(M\d)\b", m["title"]).group(1): m["id"]
                    for m in alle(f"groups/{GRUPPE}/milestones", token)
                    if re.match(r"^M\d\b", m["title"])}
    gitlab = {i["iid"]: i for i in
              alle(f"projects/{PROJEKT}/issues?state=opened", token)}

    plan: list[tuple[str, str, dict]] = []
    gespiegelt: set[int] = set()
    for f in sorted((root / "docs/issues").glob("[0-9]*.md")):
        meta, titel = frontmatter_und_titel(f)
        offen = meta.get("status") not in ("done", "rejected")
        labels = [f"priority:{meta['priority']}"]
        if STATUS_LABEL.get(meta.get("status")):
            labels.append(STATUS_LABEL[meta["status"]])
        for feld in ("host", "area"):
            if meta.get(feld):
                labels.append(f"{feld}:{meta[feld]}")
        soll = {"title": titel, "labels": sorted(labels),
                "milestone_id": meilensteine.get(meta.get("milestone")),
                "due_date": meta.get("due") or None,
                "state_event": None if offen else "close"}

        iid = int(meta["gitlab_iid"]) if meta.get("gitlab_iid") else None
        if iid is None:
            if offen:
                plan.append(("ANLEGEN", f.name,
                             {k: v for k, v in soll.items()
                              if k != "state_event" and v is not None}))
            continue
        gespiegelt.add(iid)
        ist = gitlab.get(iid)
        if ist is None:
            if offen:
                plan.append(("MELDEN", f.name,
                             {"grund": f"#{iid} auf GitLab nicht offen, "
                              f"Datei aber {meta.get('status')}"}))
            continue
        delta = {}
        if ist["title"].strip() != titel:
            delta["title"] = titel
        ist_labels = sorted(l for l in ist["labels"]
                            if l.split(":")[0] in
                            ("priority", "status", "host", "area"))
        if ist_labels != soll["labels"]:
            delta["labels"] = soll["labels"]
        if (ist.get("milestone") or {}).get("id") != soll["milestone_id"]:
            delta["milestone_id"] = soll["milestone_id"]
        if (ist.get("due_date") or None) != soll["due_date"]:
            delta["due_date"] = soll["due_date"]
        if not offen:
            delta["state_event"] = "close"
        if delta:
            plan.append(("ÄNDERN", f"{f.name} (#{iid})", delta))

    for iid in sorted(set(gitlab) - gespiegelt):
        plan.append(("MELDEN", f"management#{iid}",
                     {"grund": "offen auf GitLab ohne kanonische Datei — "
                      "wird NICHT automatisch geschlossen"}))

    modus = "AUSFÜHREN" if scharf else "DRY-RUN"
    for aktion, wer, daten in plan:
        print(f"[{modus}] {aktion} {wer}: {json.dumps(daten, ensure_ascii=False)}")
        if scharf and aktion == "ANLEGEN":
            api(f"projects/{PROJEKT}/issues", token, "POST", daten)
        elif scharf and aktion == "ÄNDERN":
            iid = int(wer.rsplit("#", 1)[1].rstrip(")"))
            api(f"projects/{PROJEKT}/issues/{iid}", token, "PUT", daten)
    print(f"spiegel_issues [{modus}]: {len(plan)} Aktionen "
          f"({sum(1 for a, _, _ in plan if a == 'MELDEN')} Meldungen)")
    return 0


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