claude-werkzeuge/.claude/skills/plane-abgleich/scripts/plane/cli.py

"""Einstiegspunkt. Hier und nur hier wird geschrieben."""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

from . import todo_datei, zustand as zustand_modul
from .abgleich import (
    Basiseintrag,
    WorkItem,
    berechne,
    item_erledigt,
)
from .api import ApiFehler, PlaneApi
from .bericht import schreibe_bericht, sichere_ausgabe
from .konfiguration import KonfigurationsFehler, lade_konfiguration


def baue_api(konfiguration) -> PlaneApi:
    return PlaneApi(konfiguration)


def sammle_projekte(quelle: Path) -> list[Path]:
    """Alle Projektordner mit einer todo.md, alphabetisch."""
    if not quelle.exists():
        return []
    return sorted(
        ordner
        for ordner in quelle.iterdir()
        if ordner.is_dir() and (ordner / "todo.md").exists()
    )


def lade_work_items(api, projekt_id: str, identifier: str) -> tuple[list[WorkItem], dict]:
    """Wandelt Plane-Rohdaten in WorkItem um. Braucht die States für die Gruppe."""
    gruppe_je_state = {s["id"]: s.get("group", "") for s in api.states(projekt_id)}
    items = []
    for roh in api.work_items(projekt_id):
        items.append(
            WorkItem(
                id=roh["id"],
                marker=f"{identifier}-{roh['sequence_id']}",
                titel=roh.get("name", ""),
                erledigt=item_erledigt(gruppe_je_state.get(roh.get("state"), "")),
            )
        )
    return items, gruppe_je_state


def _argumente(argumente: list[str]) -> argparse.Namespace:
    zerleger = argparse.ArgumentParser(prog="plane.cli", add_help=True)
    zerleger.add_argument("--anwenden", action="store_true")
    zerleger.add_argument("--nur-projekt", dest="nur_projekt", default=None)
    zerleger.add_argument("--projekte", action="store_true")
    zerleger.add_argument("--projekt-anlegen", dest="projekt_anlegen", default=None)
    zerleger.add_argument("--kennung", default=None)
    zerleger.add_argument("--konflikt", action="append", default=[])
    return zerleger.parse_args(argumente)


def hauptprogramm(argumente, umgebung, api_bauer=baue_api, strom=None) -> int:
    strom = strom if strom is not None else sys.stdout
    optionen = _argumente(argumente)

    try:
        konfiguration = lade_konfiguration(umgebung)
    except KonfigurationsFehler as fehler:
        sichere_ausgabe(strom, str(fehler))
        return 2

    api = api_bauer(konfiguration)

    try:
        if optionen.projekte:
            return _liste_projekte(api, strom)
        if optionen.projekt_anlegen:
            return _lege_projekt_an(api, optionen, strom)
        return _abgleich_lauf(api, konfiguration, optionen, strom)
    except ApiFehler as fehler:
        sichere_ausgabe(strom, str(fehler))
        return 3


def _entscheidungen(rohe: list[str]) -> dict[str, str]:
    ergebnis: dict[str, str] = {}
    for eintrag in rohe:
        marker, trenner, seite = eintrag.partition("=")
        if not trenner or seite.strip().lower() not in ("vault", "plane"):
            raise ValueError(
                f"--konflikt erwartet MARKER=vault oder MARKER=plane, "
                f"bekommen habe ich „{eintrag}“."
            )
        ergebnis[marker.strip()] = seite.strip().lower()
    return ergebnis


def _abgleich_lauf(api, konfiguration, optionen, strom) -> int:
    try:
        entscheidungen = _entscheidungen(optionen.konflikt)
    except ValueError as fehler:
        sichere_ausgabe(strom, str(fehler))
        return 2

    projekte_nach_id = {p["id"]: p for p in api.projekte()}
    alle_laeufe = []

    for ordner in sammle_projekte(konfiguration.quelle):
        if optionen.nur_projekt and optionen.nur_projekt.lower() not in ordner.name.lower():
            continue

        datei = todo_datei.lies(ordner / "todo.md")
        projekt_id = todo_datei.plane_projekt(datei.frontmatter)
        if not projekt_id:
            sichere_ausgabe(strom, "")
            sichere_ausgabe(strom, f"=== {ordner.name} ===")
            sichere_ausgabe(
                strom,
                "  UEBERSPRUNGEN - im Frontmatter der todo.md fehlt plane_project. "
                "Mit --projekte die Plane-Projekte auflisten und die UUID eintragen.",
            )
            continue

        projekt = projekte_nach_id.get(projekt_id)
        if projekt is None:
            sichere_ausgabe(strom, "")
            sichere_ausgabe(strom, f"=== {ordner.name} ===")
            sichere_ausgabe(
                strom,
                f"  UEBERSPRUNGEN - das Plane-Projekt {projekt_id} gibt es nicht "
                f"(oder der Schluessel sieht es nicht).",
            )
            continue

        items, _ = lade_work_items(api, projekt_id, projekt["identifier"])
        zustand = zustand_modul.lies(ordner / zustand_modul.DATEINAME)
        if zustand.warnung:
            sichere_ausgabe(strom, f"  HINWEIS: {zustand.warnung}")

        aktionen = berechne(datei.aufgaben, items, zustand.eintraege)
        alle_laeufe.append((ordner, datei, projekt, aktionen, zustand))
        schreibe_bericht(strom, ordner.name, aktionen, anwenden=optionen.anwenden)

    if not optionen.anwenden:
        return 0

    bekannte = {a.marker for _, _, _, aktionen, _ in alle_laeufe for a in aktionen if a.marker}
    unbekannt = sorted(set(entscheidungen) - bekannte)
    if unbekannt:
        sichere_ausgabe(
            strom,
            "Abbruch, nichts geschrieben: --konflikt nennt Marker, die in diesem Lauf "
            "nicht vorkommen: " + ", ".join(unbekannt),
        )
        return 2

    for ordner, datei, projekt, aktionen, zustand in alle_laeufe:
        _wende_an(api, projekt, ordner, datei, aktionen, zustand, entscheidungen, strom)

    return 0


def _wende_an(api, projekt, ordner, datei, aktionen, zustand, entscheidungen, strom) -> None:
    """Schritt 1: nach Plane schreiben, Zustand je Aufruf fortschreiben.
    Schritt 2: todo.md einmal komplett. Schritt 3: Laufzeit setzen."""
    from .abgleich import (
        ANHAENGEN,
        ANLEGEN,
        ITEM_ENTFERNT,
        KONFLIKT,
        NACH_PLANE,
        NACH_VAULT,
        status_fuer,
    )

    states = api.states(projekt["id"])
    zustand.plane_projekt = projekt["id"]
    aenderungen: dict[int, tuple[str, bool, str | None]] = {}
    neue_zeilen: list[str] = []

    for aktion in aktionen:
        seite = entscheidungen.get(aktion.marker or "")

        if aktion.art == ANLEGEN:
            roh = api.work_item_anlegen(
                projekt["id"],
                aktion.vault.text,
                status_fuer(states, aktion.vault.erledigt),
            )
            marker = f"{projekt['identifier']}-{roh['sequence_id']}"
            zustand.setze(
                Basiseintrag(marker, roh["id"], aktion.vault.text, aktion.vault.erledigt)
            )
            aenderungen[aktion.vault.zeilennummer] = (
                aktion.vault.text,
                aktion.vault.erledigt,
                marker,
            )

        elif aktion.art == NACH_PLANE or (aktion.art == KONFLIKT and seite == "vault"):
            api.work_item_aendern(
                projekt["id"],
                aktion.plane.id,
                {
                    "name": aktion.vault.text,
                    "state": status_fuer(states, aktion.vault.erledigt),
                },
            )
            zustand.setze(
                Basiseintrag(
                    aktion.marker, aktion.plane.id, aktion.vault.text, aktion.vault.erledigt
                )
            )

        elif aktion.art == NACH_VAULT or (aktion.art == KONFLIKT and seite == "plane"):
            aenderungen[aktion.vault.zeilennummer] = (
                aktion.plane.titel,
                aktion.plane.erledigt,
                aktion.marker,
            )
            zustand.setze(
                Basiseintrag(
                    aktion.marker, aktion.plane.id, aktion.plane.titel, aktion.plane.erledigt
                )
            )

        elif aktion.art == ANHAENGEN:
            neue_zeilen.append(
                todo_datei.baue_zeile(
                    "", aktion.plane.titel, aktion.plane.erledigt, aktion.marker
                )
            )
            zustand.setze(
                Basiseintrag(
                    aktion.marker, aktion.plane.id, aktion.plane.titel, aktion.plane.erledigt
                )
            )

        elif aktion.art == ITEM_ENTFERNT:
            aenderungen[aktion.vault.zeilennummer] = (
                aktion.vault.text,
                aktion.vault.erledigt,
                None,
            )
            zustand.entferne_marker(aktion.marker)

    if aenderungen or neue_zeilen:
        inhalt = todo_datei.schreibe(datei, aenderungen, neue_zeilen)
        (ordner / "todo.md").write_text(inhalt, encoding="utf-8")

    zustand.setze_lauf_zeit()


def _liste_projekte(api, strom) -> int:
    sichere_ausgabe(strom, "Plane-Projekte im Workspace:")
    for projekt in sorted(api.projekte(), key=lambda p: p.get("name", "")):
        sichere_ausgabe(
            strom,
            f"  {projekt.get('identifier', '?'):<12} {projekt.get('name', '')}  "
            f"{projekt['id']}",
        )
    sichere_ausgabe(
        strom,
        "Die passende UUID als plane_project ins Frontmatter der todo.md eintragen.",
    )
    return 0


def _lege_projekt_an(api, optionen, strom) -> int:
    if not optionen.kennung:
        sichere_ausgabe(
            strom, "--projekt-anlegen braucht zusaetzlich --kennung (z. B. --kennung PLA)."
        )
        return 2

    name = optionen.projekt_anlegen
    vorhanden = [p for p in api.projekte() if p.get("name", "").strip() == name.strip()]
    if vorhanden:
        sichere_ausgabe(
            strom,
            f"Abbruch: In Plane gibt es bereits ein Projekt „{name}“ "
            f"({vorhanden[0]['id']}). Nichts angelegt.",
        )
        return 2

    if not optionen.anwenden:
        sichere_ausgabe(
            strom,
            f"Probelauf - anlegen wuerde: Projekt „{name}“ "
            f"mit Kennung {optionen.kennung.upper()}. Mit --anwenden ausfuehren.",
        )
        return 0

    neu = api.projekt_anlegen(name, optionen.kennung)
    sichere_ausgabe(
        strom,
        f"Angelegt: „{name}“, Kennung {neu.get('identifier')}, "
        f"UUID {neu.get('id')}.",
    )
    sichere_ausgabe(
        strom,
        "Diese UUID als plane_project ins Frontmatter der passenden todo.md eintragen.",
    )
    return 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(hauptprogramm(sys.argv[1:], None))