management/scripts/spiegel_issues.py

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

docs/issues/ ist kanonisch (ADR-0012, seit ADR-0019 auch für die
adoptierten Komponenten-Tracker); dieses Skript bespielt die
GitLab-Ansicht, damit Board, Meilensteine und Labels weiterarbeiten.
Das Zielprojekt bestimmt das Frontmatter-Feld `projekt` (fehlt es:
management). 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. Eine offene Datei, deren
GitLab-Issue geschlossen wurde, wird wieder geöffnet.

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"
GRUPPE = "axion1337.chat"
PROJEKTE = {  # Frontmatter `projekt` -> GitLab-Projektpfad (ADR-0019)
    "management": "axion1337.chat/management",
    "gitops": "axion1337.chat/axion1337.chat-gitops",
    "threadnet-web": "axion1337.chat/ThreadNet-Web",
    "threadnet-call": "axion1337.chat/threadnet-call",
}
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"])}
    pfade = {kurz: urllib.parse.quote(pfad, safe="")
             for kurz, pfad in PROJEKTE.items()}
    gitlab = {kurz: {i["iid"]: i for i in
                     alle(f"projects/{pfade[kurz]}/issues?state=all", token)}
              for kurz in PROJEKTE}

    plan: list[tuple[str, str, dict, str]] = []
    gespiegelt: set[tuple[str, int]] = set()
    for f in sorted((root / "docs/issues").glob("[0-9]*.md")):
        meta, titel = frontmatter_und_titel(f)
        proj = meta.get("projekt") or "management"
        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}

        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 v is not None},
                             proj))
            continue
        gespiegelt.add((proj, iid))
        ist = gitlab[proj].get(iid)
        if ist is None:
            if offen:
                plan.append(("MELDEN", f.name,
                             {"grund": f"{proj}#{iid} auf GitLab nicht "
                              f"gefunden, Datei aber {meta.get('status')}"},
                             proj))
            continue
        ist_offen = ist["state"] == "opened"
        if not offen and not ist_offen:
            continue  # beidseitig zu — Historie nicht anfassen
        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 offen and not ist_offen:
            delta["state_event"] = "reopen"
        elif not offen and ist_offen:
            delta = {"state_event": "close"}
        if delta:
            plan.append(("ÄNDERN", f"{f.name} (#{iid})", delta, proj))

    for kurz in PROJEKTE:
        for iid in sorted(iid for iid, i in gitlab[kurz].items()
                          if i["state"] == "opened"
                          and (kurz, iid) not in gespiegelt):
            plan.append(("MELDEN", f"{kurz}#{iid}",
                         {"grund": "offen auf GitLab ohne kanonische Datei — "
                          "wird NICHT automatisch geschlossen"}, kurz))

    modus = "AUSFÜHREN" if scharf else "DRY-RUN"
    for aktion, wer, daten, proj in plan:
        print(f"[{modus}] {aktion} {wer}: {json.dumps(daten, ensure_ascii=False)}")
        if scharf and aktion == "ANLEGEN":
            neu = api(f"projects/{pfade[proj]}/issues", token, "POST", daten)
            # Spiegel-Adresse zurückschreiben, sonst legt der nächste
            # Lauf ein Duplikat an.
            datei = root / "docs/issues" / wer
            text = datei.read_text(encoding="utf-8")
            datei.write_text(text.replace(
                "\n---\n", f'\ngitlab_iid: "{neu["iid"]}"\n---\n', 1),
                encoding="utf-8")
            print(f"    -> {proj}#{neu['iid']} angelegt, gitlab_iid in {wer}")
        elif scharf and aktion == "ÄNDERN":
            iid = int(wer.rsplit("#", 1)[1].rstrip(")"))
            api(f"projects/{pfade[proj]}/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())