Erste Fassung: Claude-Werkzeuge aus dem Vault

.vault-tools (ohne trading-plattform-pruefung und watermarks-remover),
.claude/agents, .claude/skills, .claude/workflows, .claude/settings.json.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Pascal MichelandClaude Opus 5.5 committed 2026-10-14 11:27:59 +02:00
commit fa1c26b407
137 files changed
+43926

No files matched your search

@@ -0,0 +1,40 @@
---
name: anleitung-straffen
description: 'Baut eine bestehende technische Anleitung in den Stil aus AI/Regelwerk/schreibstil.md um — zerlegt sie in Anleitung (Schritte, Code, Prüfen) und Hintergrund (Warum, Architektur, Fallstricke), setzt eine Kommentarzeile je Codeabschnitt und belegt per Aussagenliste, dass kein Inhalt verloren ging. Nutze diesen Skill, wenn von Anleitung straffen, kürzen, umbauen, zu lang, Textwüste oder in den neuen Stil bringen die Rede ist. Nicht für neue Dokumente, die entstehen gleich im neuen Stil.'
---
# Anleitung straffen
Die Regeln stehen in `AI/Regelwerk/schreibstil.md` (Abschnitte „Anleitung und Hintergrund", „Code in Dokumenten", „Die Grenze"). Lies sie vor dem ersten Schritt. Dieser Skill beschreibt nur den Ablauf.
## Ablauf
1. **Quelle ganz lesen.** Die Anleitung vollständig, nicht in Auszügen.
2. **Aussagenliste anlegen** im Scratchpad (`aussagen.md`): jede Aussage der Quelle als eigene Zeile.
- Befehle, Pfade, Versionen, Werte, Bedingungen, Warnungen, Fehlermeldungen einzeln.
- Nummeriert: `A-01`, `A-02`, …
3. **Sortieren:** je Aussage ein Ziel.
- `anleitung` — Schritt, Codeblock, Voraussetzung, Prüfung, ⚠-Zeile am Schritt
- `hintergrund` — Warum, Architektur, Alternativen, Details zu Fallstricken, Quellen
- `streichen` — nur Floskeln ohne Aussage („Der Prozess ist nun abgeschlossen"); jede gestrichene Zeile mit Grund
4. **Dateien schreiben.**
- Dateinamen und Titel nach `schreibstil.md`, Abschnitt „Bestehende Anleitungen umbauen".
- Frontmatter der neuen Hintergrundseite: gleiches Schema; von den `bookstack_*`-Feldern nur `bookstack_shelf`, `bookstack_book`, `bookstack_chapter` übernehmen. **Nie** `bookstack_page_id`, `bookstack_hash` oder `bookstack_synced` kopieren — der Upload-Skill würde sonst die Seite der Anleitung mit dem Hintergrund überschreiben.
- Links zwischen beiden Seiten nach `schreibstil.md`, Abschnitt „Anleitung und Hintergrund".
- Codeblöcke: Zeilenkommentare entfernen, je Abschnitt eine Kommentarzeile; was die Zeilenkommentare erklärten, wandert in den Hintergrund.
5. **Verlustprobe.** Tabelle `Aussage | Ziel | Fundstelle (Datei, Überschrift)` für jede `A-NN`.
- Fehlt eine Aussage in beiden Dateien und steht nicht unter `streichen`: nachtragen, bevor du weitermachst.
- Bedingungen und Versionsangaben wörtlich gegen die Quelle vergleichen, nicht aus dem Gedächtnis.
6. **Nacharbeiten.**
- Frontmatter: `updated` auf heute, Tag `review` in `tags`.
- **Im Text** beider Seiten ein `#review` setzen und ein vorhandenes `#review-done` entfernen. Der Upload-Skill liest nur den Text (`bookstack-upload/scripts/bookstack/auswahl.py`); ein Frontmatter-Tag allein hält eine freigegebene Seite nicht vom Upload ab.
- Nach Pascals Freigabe: erst die Hintergrundseite hochladen, dann die Anleitung ein zweites Mal. Beim ersten Lauf hat der Hintergrund noch keine Seiten-ID, der Link in der Anleitung wird sonst zu reinem Text.
- Seed-MOC des Themenordners: Eintrag für die Hintergrundseite ergänzen.
- Linkcheck: `PYTHONIOENCODING=utf-8 python .vault-tools/linkcheck/linkcheck.py --vault .` ohne neue tote Links.
7. **Bericht an Pascal:** beide Pfade, Zahl der Aussagen je Ziel, jede gestrichene Aussage mit Grund, alles, was du nicht zuordnen konntest.
## Grenzen
- Ändert keine Aussage, nur ihre Form und ihren Ort. Findest du einen inhaltlichen Fehler, melde ihn und lass ihn stehen.
- Regelteile von ISMS- und Compliance-Dokumenten fasst der Skill nicht an (siehe „Wo der Stil nicht gilt").
- Mehrere Anleitungen: je Anleitung ein eigener Durchgang mit eigener Aussagenliste.
+77
View File
@@ -0,0 +1,77 @@
---
name: bookstack-upload
description: 'Lädt freigegebene Vault-Notizen (Tag #review-done) aus dem Ordner /Bookstack über die BookStack-API hoch — legt Seiten an oder aktualisiert sie, setzt Buch- und Kapitelbeschreibungen, zieht Verweise zwischen den Seiten nach, löst Regal/Buch/Kapitel aus dem Frontmatter auf und fragt nach, wenn die Zuordnung fehlt. Nutze diesen Skill, wenn von BookStack-Upload, Wiki-Upload, „hochladen nach Bookstack", „Doku veröffentlichen" oder vom Verarbeiten des /Bookstack-Ordners die Rede ist.'
---
# BookStack-Upload
Spielt freigegebene Dokumentation aus dem Vault in die BookStack-Instanz.
## Voraussetzungen prüfen
Setze `BOOKSTACK_URL`, `BOOKSTACK_TOKEN_ID` und `BOOKSTACK_TOKEN_SECRET` voraus. Sind sie
nicht gesetzt, meldet das Skript das verständlich — gib die Meldung weiter und brich ab.
Frage niemals nach dem Token und schreibe ihn nirgends hin.
## Ablauf
1. **Immer zuerst der Probelauf.** Ohne `--anwenden` schreibt nichts:
```
python -m bookstack.hochladen
```
Arbeitsverzeichnis: `D:\Claude\Code Projekte\active - Bookstack Upload\skill\scripts`
2. **Bericht lesen.** Er listet je Datei ANLEGEN, AKTUALISIEREN, ÜBERSPRUNGEN oder FEHLER,
bei `_Buchbeschreibung.md` und `_Kapitelbeschreibung.md` zusätzlich BESCHREIBUNG
(„würde gesetzt“ oder „unverändert“). Die Zeile
`N× Verweis wird in der Verweisrunde aufgelöst` zählt Links auf Seiten, die erst im
selben Lauf entstehen, und ist kein Fehler. FEHLER-Zeilen mit „Beschreibung zu lang“
oder „Vorspann fehlt“ dem Anwender vor dem Upload vorlegen: Er kürzt oder ergänzt den
Vorspann (Text zwischen H1 und erster H2). Ebenso FEHLER-Zeilen mit
„Beschreibung nicht wohlgeformt: Hervorhebung falsch verschachtelt, nichts gesendet“:
Er korrigiert die Hervorhebung im Vorspann, etwa `***a***` → `**a**`.
3. **Fehlende Zuordnung klären.** Meldet der Bericht für eine Notiz, dass Buch oder
Kapitel nicht existiert oder `bookstack_book` fehlt:
- Vorhandene Struktur zeigen: `python -m bookstack.hochladen --baum`
- Dem Anwender die Regale und Bücher aus dieser Ausgabe zur Auswahl stellen und
fragen, wo die Notiz landen soll. Nicht raten.
- Die Antwort ins Frontmatter der Notiz schreiben (`bookstack_shelf`,
`bookstack_book`, `bookstack_chapter`).
- Soll etwas Neues angelegt werden, den Anwender das ausdrücklich bestätigen lassen
und dann `--struktur-anlegen` ergänzen.
- Meldet der Bericht bei einer Beschreibungsdatei „bookstack_chapter fehlt in der
Kapitelbeschreibung“ oder „bookstack_chapter gehört nicht in eine Buchbeschreibung“,
genauso nachfragen und das Frontmatter korrigieren.
4. **Bericht vorlegen und Freigabe einholen.** Erst nach ausdrücklichem „ja" schreiben:
```
python -m bookstack.hochladen --anwenden
```
5. **Ergebnis melden.** Was wurde angelegt, was aktualisiert, was übersprungen. Hinweise
zu nicht übertragenen Bildern und nicht auflösbaren Wikilinks (offene Verweise)
weitergeben. `ARCHIVIERT`-Zeilen sind normaler Ablauf (Notizen wandern nach
erfolgreichem Upload nach `nicht hochladen - ARCHIV/`), kein Fehler. Ebenso
`VERLINKT`-Zeilen: Die Verweisrunde hat Links auf Seiten desselben Laufs nachgezogen.
`ARCHIV-FEHLER` dagegen melden: Im Archiv liegt schon eine Datei mit demselben Pfad,
die Notiz ist liegen geblieben. Nicht selbst löschen, den Anwender entscheiden lassen.
## Grenzen
- `--anwenden` niemals ungefragt ausführen. Der Upload ist nach außen sichtbar.
- Notizen mit offenem `#review` werden bewusst übersprungen. Nicht umgehen.
- Bilder werden in die BookStack-Bildergalerie hochgeladen und eingebunden. Nicht
gefundene oder abgelehnte Bilder erscheinen als Hinweis im Bericht.
- Wikilinks auf bereits hochgeladene Notizen und auf Seiten desselben Laufs werden zu
echten Seitenlinks, Links auf Beschreibungsdateien zeigen auf deren Übersichtsseite.
Verweise auf Notizen ohne Seite bleiben Klartext und werden im Bericht mit dem Ziel
gemeldet, wie es in der Notiz steht.
- `_Buchbeschreibung.md` und `_Kapitelbeschreibung.md` brauchen `#review-done` wie jede
Seite. Andere Dateien mit `_` oder `.` am Anfang lädt der Lauf ohne `--nur` nicht hoch.
- Nach dem Archivieren zeigen Wikilinks aus dem übrigen Vault noch auf den alten Pfad.
Der Skill zieht sie nicht nach; das beim Ergebnis dem Anwender sagen.
Whitespace-only changes.
@@ -0,0 +1,50 @@
"""Entscheidet je Notiz, ob sie angelegt, aktualisiert oder übersprungen wird."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from enum import Enum
from bookstack.notiz import Notiz
class Aktion(Enum):
ANLEGEN = "anlegen"
AKTUALISIEREN = "aktualisieren"
UEBERSPRINGEN = "überspringen"
@dataclass(frozen=True)
class Entscheidung:
aktion: Aktion
seiten_id: int | None
pruefsumme: str
begruendung: str
def berechne_pruefsumme(
titel: str, markdown: str, buch_id: int, kapitel_id: int | None
) -> str:
"""Deckt Inhalt *und* Ablageort ab — ein verschobenes Ziel gilt als Änderung."""
roh = "\x00".join([titel, str(buch_id), str(kapitel_id or 0), markdown])
return "sha256:" + hashlib.sha256(roh.encode("utf-8")).hexdigest()
def entscheide(notiz: Notiz, pruefsumme: str) -> Entscheidung:
roh_id = notiz.frontmatter.get("bookstack_page_id", "").strip()
if not roh_id:
return Entscheidung(Aktion.ANLEGEN, None, pruefsumme, "noch nie hochgeladen")
try:
seiten_id = int(roh_id)
except ValueError:
return Entscheidung(
Aktion.ANLEGEN, None, pruefsumme, f"bookstack_page_id unlesbar: {roh_id!r}"
)
if notiz.frontmatter.get("bookstack_hash", "").strip() == pruefsumme:
return Entscheidung(
Aktion.UEBERSPRINGEN, seiten_id, pruefsumme, "unverändert seit letztem Upload"
)
return Entscheidung(Aktion.AKTUALISIEREN, seiten_id, pruefsumme, "Inhalt hat sich geändert")
@@ -0,0 +1,142 @@
"""Schmaler HTTP-Client für die BookStack-API — Standardbibliothek, kein requests."""
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
import uuid
from collections.abc import Callable
SEITENGROESSE = 500
WIEDERHOLUNGEN_BEI_429 = 2
# Ohne eigene Kennung schickt urllib "Python-urllib/3.x". Cloudflare und ähnliche
# Schutzschichten vor BookStack-Instanzen beantworten das mit 403, noch bevor die
# API das Token überhaupt sieht. Gefunden beim ersten Lauf gegen die echte Instanz.
KENNUNG = "bookstack-upload/1.0 (Claude Code Skill)"
class ApiFehler(Exception):
def __init__(self, status: int, meldung: str, pfad: str) -> None:
super().__init__(f"BookStack antwortete {status} auf {pfad}: {meldung}")
self.status = status
self.meldung = meldung
self.pfad = pfad
def _baue_multipart(
felder: dict[str, str], datei: tuple[str, str, bytes, str]
) -> tuple[bytes, str]:
"""Baut einen multipart/form-data-Körper von Hand — die Standardbibliothek kann das nicht."""
grenze = "----bookstackupload" + uuid.uuid4().hex
teile: list[bytes] = []
for name, wert in felder.items():
teile.append(
f'--{grenze}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{wert}\r\n'.encode(
"utf-8"
)
)
feldname, dateiname, inhalt, mime = datei
teile.append(
(
f"--{grenze}\r\n"
f'Content-Disposition: form-data; name="{feldname}"; filename="{dateiname}"\r\n'
f"Content-Type: {mime}\r\n\r\n"
).encode("utf-8")
+ inhalt
+ b"\r\n"
)
teile.append(f"--{grenze}--\r\n".encode("utf-8"))
return b"".join(teile), f"multipart/form-data; boundary={grenze}"
class BookStackApi:
def __init__(
self,
basis_url: str,
token_id: str,
token_secret: str,
*,
mindestabstand: float = 0.34,
schlaefer: Callable[[float], None] = time.sleep,
uhr: Callable[[], float] = time.monotonic,
) -> None:
self.basis_url = basis_url.rstrip("/")
self._autorisierung = f"Token {token_id}:{token_secret}"
self._mindestabstand = mindestabstand
self._schlaefer = schlaefer
self._uhr = uhr
self._letzter_aufruf: float | None = None
def _drossle(self) -> None:
jetzt = self._uhr()
if self._letzter_aufruf is not None:
rest = self._mindestabstand - (jetzt - self._letzter_aufruf)
if rest > 0:
self._schlaefer(rest)
self._letzter_aufruf = self._uhr()
def _einmal(self, methode: str, pfad: str, koerper: bytes | None, inhaltstyp: str | None) -> dict:
self._drossle()
anfrage = urllib.request.Request(self.basis_url + pfad, data=koerper, method=methode)
anfrage.add_header("Authorization", self._autorisierung)
anfrage.add_header("Accept", "application/json")
anfrage.add_header("User-Agent", KENNUNG)
if inhaltstyp is not None:
anfrage.add_header("Content-Type", inhaltstyp)
try:
with urllib.request.urlopen(anfrage, timeout=60) as antwort:
roh = antwort.read()
except urllib.error.HTTPError as fehler:
roh_fehler = fehler.read()
try:
meldung = json.loads(roh_fehler)["error"]["message"]
except (ValueError, KeyError, TypeError):
meldung = str(fehler.reason)
raise ApiFehler(fehler.code, meldung, pfad) from None
except urllib.error.URLError as fehler:
raise ApiFehler(0, f"nicht erreichbar ({fehler.reason})", pfad) from None
return json.loads(roh) if roh else {}
def _mit_wiederholung(
self, methode: str, pfad: str, koerper: bytes | None, inhaltstyp: str | None
) -> dict:
for versuch in range(WIEDERHOLUNGEN_BEI_429 + 1):
try:
return self._einmal(methode, pfad, koerper, inhaltstyp)
except ApiFehler as fehler:
if fehler.status != 429 or versuch == WIEDERHOLUNGEN_BEI_429:
raise
self._schlaefer(2.0 * (versuch + 1))
raise AssertionError("unerreichbar")
def anfrage(self, methode: str, pfad: str, daten: dict | None = None) -> dict:
"""Ein JSON-API-Aufruf. Bei 429 wird bis zu zweimal nachgefasst."""
if daten is None:
return self._mit_wiederholung(methode, pfad, None, None)
return self._mit_wiederholung(
methode, pfad, json.dumps(daten).encode("utf-8"), "application/json"
)
def sende_formular(
self, pfad: str, felder: dict[str, str], datei: tuple[str, str, bytes, str]
) -> dict:
"""POST als multipart/form-data — nötig für die Bildergalerie."""
koerper, inhaltstyp = _baue_multipart(felder, datei)
return self._mit_wiederholung("POST", pfad, koerper, inhaltstyp)
def liste(self, pfad: str) -> list[dict]:
"""Holt alle Seiten einer Sammlung und gibt sie als eine Liste zurück."""
eintraege: list[dict] = []
versatz = 0
trenner = "&" if "?" in pfad else "?"
while True:
antwort = self.anfrage(
"GET", f"{pfad}{trenner}count={SEITENGROESSE}&offset={versatz}"
)
teil = antwort.get("data", [])
eintraege.extend(teil)
versatz += len(teil)
if not teil or versatz >= int(antwort.get("total", len(eintraege))):
return eintraege
@@ -0,0 +1,46 @@
"""Verschiebt bereits hochgeladene Notizen aus dem aktiven Bereich in ein Archiv.
Ersteller: Pascal Michel
Erstellt: 2026-08-11
Doku: docs/code/archiv.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
from pathlib import Path
ARCHIV_ORDNERNAME = "nicht hochladen - ARCHIV"
def ist_archiviert(pfad: Path, quelle: Path) -> bool:
"""True, wenn der Pfad bereits unterhalb des Archiv-Unterverzeichnisses liegt.
Liegt pfad außerhalb von quelle (z. B. --nur mit einem externen Pfad), gilt
das als „nicht archiviert" — archiviere() wird für solche Pfade gar nicht
erst aufgerufen (siehe hochladen.py).
"""
try:
teile = pfad.relative_to(quelle).parts
except ValueError:
return False
return ARCHIV_ORDNERNAME in teile
def archivziel(pfad: Path, quelle: Path) -> Path:
"""Berechnet den Zielpfad im Archiv, ohne etwas zu verschieben."""
relativ = pfad.relative_to(quelle)
return quelle / ARCHIV_ORDNERNAME / relativ
def archiviere(pfad: Path, quelle: Path) -> Path:
"""Verschiebt die Datei ins Archiv-Unterverzeichnis; legt Ordner bei Bedarf an.
Existiert das Ziel schon, bleibt die Quelle liegen und es kommt FileExistsError.
Unter Windows wirft rename das ohnehin, unter POSIX würde es still überschreiben.
"""
ziel = archivziel(pfad, quelle)
if ziel.exists():
raise FileExistsError(f"Archivziel existiert schon: {ziel}")
ziel.parent.mkdir(parents=True, exist_ok=True)
pfad.rename(ziel)
return ziel
@@ -0,0 +1,76 @@
"""Findet im Quellordner die Notizen und Beschreibungsdateien, die zum Upload freigegeben sind.
Ersteller: Pascal Michel
Erstellt: 2026-08-05
Doku: docs/code/auswahl.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
import re
from pathlib import Path
from bookstack.notiz import Notiz, lade_notiz
MAX_TITEL = 255
FREIGABE = re.compile(r"(?<![\w#-])#review-done(?![\w-])")
OFFEN = re.compile(r"(?<![\w#-])#review(?![\w-])")
UEBERSCHRIFT = re.compile(r"^#\s+(.+)$", re.MULTILINE)
BESCHREIBUNGSDATEIEN: dict[str, str] = {
"_Buchbeschreibung.md": "buch",
"_Kapitelbeschreibung.md": "kapitel",
}
def ist_freigegeben(notiz: Notiz) -> bool:
"""Freigegeben heißt: #review-done vorhanden und kein offenes #review mehr."""
return bool(FREIGABE.search(notiz.roh)) and not OFFEN.search(notiz.roh)
def beschreibungsart(pfad: Path) -> str | None:
"""'buch', 'kapitel' oder None — exakter Vergleich auf pfad.name."""
return BESCHREIBUNGSDATEIEN.get(pfad.name)
def finde_kandidaten(quelle: Path) -> list[Notiz]:
if not quelle.is_dir():
return []
gefunden: list[Notiz] = []
for pfad in sorted(quelle.rglob("*.md")):
if pfad.name.startswith((".", "_")):
continue
notiz = lade_notiz(pfad)
if ist_freigegeben(notiz):
gefunden.append(notiz)
return gefunden
def finde_beschreibungen(quelle: Path) -> list[Notiz]:
"""Freigegebene Beschreibungsdateien unter quelle (rglob, Archiv eingeschlossen).
Reihenfolge: alle 'buch' vor allen 'kapitel', je Art nach Pfad sortiert."""
if not quelle.is_dir():
return []
buecher: list[Notiz] = []
kapitel: list[Notiz] = []
for pfad in sorted(quelle.rglob("*.md")):
art = beschreibungsart(pfad)
if art is None:
continue
notiz = lade_notiz(pfad)
if not ist_freigegeben(notiz):
continue
if art == "buch":
buecher.append(notiz)
else:
kapitel.append(notiz)
return buecher + kapitel
def titel_von(notiz: Notiz) -> str:
"""Erste H1, sonst Frontmatter-title, sonst Dateiname."""
treffer = UEBERSCHRIFT.search(notiz.rumpf)
if treffer:
titel = treffer.group(1).strip()
else:
titel = notiz.frontmatter.get("title", "").strip() or notiz.pfad.stem
return titel[:MAX_TITEL]
@@ -0,0 +1,296 @@
"""Wandelt den Vorspann einer Beschreibungsdatei in Beschreibungs-HTML und setzt es per PUT an Buch oder Kapitel.
Ersteller: Pascal Michel
Erstellt: 2026-09-30
Doku: docs/code/beschreibung.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
import hashlib
import re
from collections.abc import Mapping
from dataclasses import dataclass, field
from html import escape
from html.parser import HTMLParser
from bookstack.api import BookStackApi
from bookstack.konvertierung import EINBETTUNG, REVIEW_TAG, WIKILINK, ankertext, zerlege_wikilink
from bookstack.verweise import Verweisindex
MAX_BESCHREIBUNG_HTML = 2000
UEBERSICHTSTITEL: dict[str, str] = {"buch": "Über dieses Buch", "kapitel": "Über dieses Kapitel"}
ZAUNMARKEN = ("```", "~~~")
H1_ZEILE = re.compile(r"^#\s+\S")
H2_ZEILE = re.compile(r"^##\s")
UL_ANFANG = re.compile(r"^[-*+]\s")
UL_PUNKT = re.compile(r"^\s*[-*+]\s+(.*)$")
OL_ANFANG = re.compile(r"^\d+[.)]\s")
OL_PUNKT = re.compile(r"^\s*\d+[.)]\s+(.*)$")
KLEINE_UEBERSCHRIFT = re.compile(r"^#{3,6}\s+(.*)$")
ZITAT = re.compile(r"^\s*>\s?")
CALLOUT_ZEILE = re.compile(r"^\[!(\w+)\][ \t]*(.*)$")
CODE_SPAN = re.compile(r"`([^`\n]+)`")
MARKDOWN_LINK = re.compile(r"\[([^\]\n]+)\]\(([^)\s]+)\)")
ERLAUBTE_LINKANFAENGE = ("http://", "https://", "mailto:", "/")
FETT = re.compile(r"\*\*(?=\S)(.+?)(?<=\S)\*\*")
KURSIV = re.compile(r"\*(?=\S)([^*]+?)(?<=\S)\*")
PLATZHALTER = re.compile(r"\x00(\d+)\x00")
class BeschreibungsFehler(Exception):
"""Beschreibung nicht setzbar; Text geht als FEHLER in den Bericht."""
@dataclass
class BeschreibungsHtml:
html: str
aufgeloeste_verweise: dict[str, str] = field(default_factory=dict)
offene_verweise: list[str] = field(default_factory=list)
def pruefe_ziel_felder(art: str, frontmatter: Mapping[str, str]) -> None:
"""Buchbeschreibung ohne bookstack_chapter, Kapitelbeschreibung mit nicht leerem bookstack_chapter."""
kapitel = (frontmatter.get("bookstack_chapter") or "").strip()
if art == "buch" and kapitel:
raise BeschreibungsFehler("bookstack_chapter gehört nicht in eine Buchbeschreibung")
if art != "buch" and not kapitel:
raise BeschreibungsFehler("bookstack_chapter fehlt in der Kapitelbeschreibung")
def _zaunmarke(zeile: str) -> str | None:
anfang = zeile.lstrip()[:3]
return anfang if anfang in ZAUNMARKEN else None
def vorspann(rumpf: str) -> str:
"""Text nach der ersten H1 bis vor die erste folgende H2, gestrippt.
Ohne H1 ab dem Anfang, ohne H2 bis zum Ende. Zeilen in Code-Zäunen zählen nicht als Überschrift.
"""
zeilen = rumpf.split("\n")
ausserhalb: list[bool] = []
zaun: str | None = None
for zeile in zeilen:
marke = _zaunmarke(zeile)
if zaun is None:
ausserhalb.append(marke is None)
zaun = marke
else:
ausserhalb.append(False)
if marke == zaun:
zaun = None
anfang = next(
(nummer + 1 for nummer, zeile in enumerate(zeilen) if ausserhalb[nummer] and H1_ZEILE.match(zeile)),
0,
)
ende = next(
(
nummer
for nummer in range(anfang, len(zeilen))
if ausserhalb[nummer] and H2_ZEILE.match(zeilen[nummer])
),
len(zeilen),
)
return "\n".join(zeilen[anfang:ende]).strip()
def _bloecke(text: str) -> list[tuple[str, list[str]]]:
"""Textblöcke an Leerzeilen getrennt; Code-Zäune als eigene Blöcke ohne die Zaunzeilen."""
bloecke: list[tuple[str, list[str]]] = []
laufend: list[str] = []
code: list[str] = []
zaun: str | None = None
for zeile in text.split("\n"):
marke = _zaunmarke(zeile)
if zaun is not None:
if marke == zaun:
bloecke.append(("code", code))
code, zaun = [], None
else:
code.append(zeile)
continue
if marke is not None:
if laufend:
bloecke.append(("text", laufend))
laufend = []
zaun = marke
elif zeile.strip():
laufend.append(zeile)
elif laufend:
bloecke.append(("text", laufend))
laufend = []
if zaun is not None:
bloecke.append(("code", code))
if laufend:
bloecke.append(("text", laufend))
return bloecke
def _inline(text: str, verweise: Verweisindex, ergebnis: BeschreibungsHtml) -> str:
"""Inline-Regeln 8.1 bis 8.6; Code, Einbettungen und Links stehen bis zum Schluss als Platzhalter."""
stuecke: list[str] = []
def merke(stueck: str) -> str:
stuecke.append(stueck)
return f"\x00{len(stuecke) - 1}\x00"
def code_span(treffer: re.Match[str]) -> str:
return merke(f"<code>{escape(treffer.group(1), quote=False)}</code>")
def einbettung(treffer: re.Match[str]) -> str:
name = treffer.group(1).strip()
if name.endswith("\\"):
name = name[:-1].strip()
return merke(escape(name, quote=False))
def wikilink(treffer: re.Match[str]) -> str:
anker = ankertext(treffer.group(1), treffer.group(2))
if anker is not None:
return merke(escape(anker, quote=False))
ziel, anzeigetext = zerlege_wikilink(treffer.group(1), treffer.group(2))
url = verweise.aufloesen(ziel)
if url:
ergebnis.aufgeloeste_verweise[ziel] = url
return merke(f'<a href="{escape(url, quote=True)}">{escape(anzeigetext, quote=False)}</a>')
if ziel not in ergebnis.offene_verweise:
ergebnis.offene_verweise.append(ziel)
return merke(escape(anzeigetext, quote=False))
def markdown_link(treffer: re.Match[str]) -> str:
linktext, url = treffer.group(1), treffer.group(2)
if not url.startswith(ERLAUBTE_LINKANFAENGE):
return treffer.group(0)
return merke(f'<a href="{escape(url, quote=True)}">{escape(linktext, quote=False)}</a>')
text = CODE_SPAN.sub(code_span, text)
text = EINBETTUNG.sub(einbettung, text)
text = WIKILINK.sub(wikilink, text)
text = MARKDOWN_LINK.sub(markdown_link, text)
text = escape(text, quote=False)
text = FETT.sub(r"<strong>\1</strong>", text)
text = KURSIV.sub(r"<em>\1</em>", text)
# Ein Linktext kann selbst einen Platzhalter tragen (Code im Alias); deshalb bis keiner mehr übrig ist.
while PLATZHALTER.search(text):
text = PLATZHALTER.sub(lambda treffer: stuecke[int(treffer.group(1))], text)
return text
def _liste(
tag: str,
punkt: re.Pattern[str],
zeilen: list[str],
verweise: Verweisindex,
ergebnis: BeschreibungsHtml,
) -> str:
eintraege: list[str] = []
for zeile in zeilen:
treffer = punkt.match(zeile)
if treffer:
eintraege.append(treffer.group(1).strip())
else:
eintraege[-1] = f"{eintraege[-1]} {zeile.strip()}"
innen = "".join(f"<li>{_inline(eintrag, verweise, ergebnis)}</li>" for eintrag in eintraege)
return f"<{tag}>{innen}</{tag}>"
def _absaetze(zeilen: list[str], verweise: Verweisindex, ergebnis: BeschreibungsHtml) -> list[str]:
teile: list[str] = []
laufend: list[str] = []
def abschliessen() -> None:
if laufend:
teile.append(f"<p>{_inline(' '.join(laufend), verweise, ergebnis)}</p>")
laufend.clear()
for roh in zeilen:
zeile = ZITAT.sub("", roh, count=1).strip()
callout = CALLOUT_ZEILE.match(zeile)
if callout:
zeile = f"**{callout.group(2).strip() or callout.group(1).capitalize()}**"
ueberschrift = KLEINE_UEBERSCHRIFT.match(zeile)
if ueberschrift:
abschliessen()
inhalt = _inline(ueberschrift.group(1).strip(), verweise, ergebnis)
teile.append(f"<p><strong>{inhalt}</strong></p>")
elif zeile:
laufend.append(zeile)
else:
abschliessen()
abschliessen()
return teile
def zu_beschreibungs_html(markdown: str, verweise: Verweisindex) -> BeschreibungsHtml:
"""Markdown des Vorspanns als HTML aus p, a href, ul, ol, li, strong, em und code."""
ergebnis = BeschreibungsHtml(html="")
text = REVIEW_TAG.sub("", markdown.replace("\x00", ""))
text = "\n".join(zeile.rstrip() for zeile in text.split("\n"))
teile: list[str] = []
for art, zeilen in _bloecke(text):
if art == "code":
inhalt = " ".join(zeile.strip() for zeile in zeilen if zeile.strip())
if inhalt:
teile.append(f"<p><code>{escape(inhalt, quote=False)}</code></p>")
elif UL_ANFANG.match(zeilen[0]):
teile.append(_liste("ul", UL_PUNKT, zeilen, verweise, ergebnis))
elif OL_ANFANG.match(zeilen[0]):
teile.append(_liste("ol", OL_PUNKT, zeilen, verweise, ergebnis))
else:
teile.extend(_absaetze(zeilen, verweise, ergebnis))
ergebnis.html = "".join(teile)
return ergebnis
class _Verschachtelung(HTMLParser):
"""Merkt sich, ob jedes schließende Tag das zuletzt geöffnete schließt."""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.offen: list[str] = []
self.fehler = False
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
if tag != "br":
self.offen.append(tag)
def handle_endtag(self, tag: str) -> None:
if tag == "br":
return
if self.offen and self.offen[-1] == tag:
self.offen.pop()
else:
self.fehler = True
def pruefe_beschreibung_html(html: str) -> None:
"""Leer, über MAX_BESCHREIBUNG_HTML Zeichen oder falsch verschachtelt: BeschreibungsFehler, gesendet wird dann nichts."""
if not html.strip():
raise BeschreibungsFehler("Vorspann fehlt")
if len(html) > MAX_BESCHREIBUNG_HTML:
raise BeschreibungsFehler(
f"Beschreibung zu lang: {len(html)} Zeichen HTML, erlaubt {MAX_BESCHREIBUNG_HTML}, nichts gesendet"
)
# Der Server repariert falsch verschachteltes HTML nach eigenem Ermessen und speichert dann
# anderes HTML als gesendet. Die Prüfsumme läuft über das gesendete HTML und merkt das nicht.
pruefer = _Verschachtelung()
pruefer.feed(html)
pruefer.close()
if pruefer.fehler or pruefer.offen:
raise BeschreibungsFehler(
"Beschreibung nicht wohlgeformt: Hervorhebung falsch verschachtelt, nichts gesendet"
)
def beschreibungs_pruefsumme(ziel_id: int, html: str) -> str:
"""Prüfsumme über Ziel-ID und gesendetes HTML; ein anderes Ziel gilt als Änderung."""
return "sha256:" + hashlib.sha256(f"{ziel_id}\x00{html}".encode("utf-8")).hexdigest()
def setze_beschreibung(api: BookStackApi, art: str, ziel_id: int, html: str) -> None:
"""PUT description_html an das Buch (art 'buch') oder das Kapitel; ID 0 ist ein Platzhalter und wird nie gesendet."""
if ziel_id == 0:
raise BeschreibungsFehler("Beschreibung nicht gesetzt: Ziel-ID 0 ist ein Platzhalter aus dem Probelauf")
pfad = f"/api/books/{ziel_id}" if art == "buch" else f"/api/chapters/{ziel_id}"
api.anfrage("PUT", pfad, {"description_html": html})
@@ -0,0 +1,125 @@
"""Findet Bilder im Vault und lädt sie in die BookStack-Bildergalerie.
Ersteller: Pascal Michel
Erstellt: 2026-08-05
Doku: docs/code/bilder.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
import os
from pathlib import Path
from bookstack.api import ApiFehler, BookStackApi
# In D:\Claude\skills\ liegt ein grosses Fremd-Repo — laut CLAUDE.md haengt sich
# ein rekursiver Lauf dort praktisch endlos auf. Diese Ordner werden uebersprungen.
UEBERSPRINGEN = {"skills", ".git", ".venv", "node_modules", ".obsidian", "__pycache__", ".pytest_cache"}
TYPEN = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
".avif": "image/avif",
}
# Die API-Doku nennt „max 50KB"; das ist mit hoher Wahrscheinlichkeit die
# Laravel-Regel max:50000, die bei Dateien in Kilobyte rechnet (~48,8 MB).
# Deshalb hier eine Grenze knapp darunter — die echte Ablehnung der API wird
# ohnehin woertlich weitergereicht.
MAX_BYTES = 48 * 1024 * 1024
class BildFehler(Exception):
"""Ein Bild konnte nicht übertragen werden."""
def finde_bilddatei(vault: Path, notiz_pfad: Path, name: str) -> Path | None:
"""Sucht erst neben der Notiz, dann in der Vault-Wurzel, dann im ganzen Vault."""
for kandidat in (notiz_pfad.parent / name, vault / name):
if kandidat.is_file():
return kandidat
for wurzel, ordner, dateien in os.walk(vault):
ordner[:] = [eintrag for eintrag in ordner if eintrag not in UEBERSPRINGEN]
if name in dateien:
return Path(wurzel) / name
return None
class Bilduploader:
def __init__(self, api: BookStackApi, vault: Path) -> None:
self._api = api
self._vault = vault
def _lies_galerie(self, seiten_id: int) -> dict[str, str]:
"""Bereits an dieser Seite hängende Galeriebilder, Name → URL; ApiFehler geht durch."""
eintraege = self._api.liste(f"/api/image-gallery?filter[uploaded_to]={seiten_id}")
return {str(eintrag.get("name", "")): str(eintrag.get("url", "")) for eintrag in eintraege}
def _vorhandene(self, seiten_id: int) -> dict[str, str]:
"""Wie _lies_galerie, aber ein ApiFehler ergibt {}; nur für lade_hoch."""
try:
return self._lies_galerie(seiten_id)
except ApiFehler:
return {}
def vorhandene(self, seiten_id: int) -> dict[str, str]:
"""Galeriebilder der Seite, Name → URL; nur lesend, für die Verweisrunde.
Ein ApiFehler geht durch: Ein leeres Ergebnis sähe dort aus wie ein fehlendes Bild.
"""
return self._lies_galerie(seiten_id)
def lade_hoch(
self, notiz_pfad: Path, seiten_id: int, namen: list[str]
) -> tuple[dict[str, str], list[str]]:
"""Liefert (Name → URL) und eine Liste von Hinweisen für den Bericht."""
tabelle: dict[str, str] = {}
hinweise: list[str] = []
if not namen:
return tabelle, hinweise
# Ein Lesefehler der Galerie ergibt hier {}, gewollt: Das Bild wird neu hochgeladen, statt den
# Upload scheitern zu lassen, und kostet höchstens ein Doppelbild. Die Verweisrunde liest über
# vorhandene, dort geht der Fehler durch.
vorhanden = self._vorhandene(seiten_id)
for name in dict.fromkeys(namen):
if name in vorhanden and vorhanden[name]:
tabelle[name] = vorhanden[name]
continue
pfad = finde_bilddatei(self._vault, notiz_pfad, name)
if pfad is None:
hinweise.append(f"Bild nicht gefunden im Vault: {name}")
continue
mime = TYPEN.get(pfad.suffix.lower())
if mime is None:
hinweise.append(f"Bildtyp nicht unterstützt: {name}")
continue
groesse = pfad.stat().st_size
if groesse > MAX_BYTES:
hinweise.append(f"Bild zu groß ({groesse // (1024 * 1024)} MB): {name}")
continue
try:
antwort = self._api.sende_formular(
"/api/image-gallery",
{"type": "gallery", "uploaded_to": str(seiten_id), "name": name},
("image", name, pfad.read_bytes(), mime),
)
except ApiFehler as fehler:
hinweise.append(f"Bild abgelehnt ({name}): {fehler.meldung}")
continue
url = str(antwort.get("url", ""))
if url:
tabelle[name] = url
else:
hinweise.append(f"Bild ohne URL zurückgekommen: {name}")
return tabelle, hinweise
@@ -0,0 +1,69 @@
"""Ergebnistypen eines Laufs: Berichtszeile samt dem Stand, den die Verweisrunde braucht.
Ersteller: Pascal Michel
Erstellt: 2026-09-30
Doku: docs/code/ergebnis.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from bookstack.struktur import Ziel
# Aktionen, deren Datei danach archiviert wird und deren Seite in den Endindex kommt.
ARCHIVIERBAR = ("ANLEGEN", "AKTUALISIEREN", "ÜBERSPRUNGEN")
@dataclass
class Seitenstand:
"""Was Durchgang 1 über eine Seite weiß; Eingang der Verweisrunde."""
titel: str
rumpf: str # ohne Titelüberschrift, Eingang für konvertiere
ziel: Ziel
seiten_id: int | None # None nur bei ANLEGEN im Probelauf
prioritaet: int | None
markdown: str # bildfrei, Verweise nach Index aus Durchgang 1
aufgeloeste_verweise: dict[str, str]
offene_verweise: list[str]
bildnamen: list[str]
bilder: dict[str, str] | None # Tabelle des letzten PUT in Durchgang 1; None = kein PUT
@dataclass
class Beschreibungsstand:
"""Was Durchgang 1 über eine Buch- oder Kapitelbeschreibung weiß."""
art: str # "buch" | "kapitel"
ziel_id: int
vorspann: str
html: str
aufgeloeste_verweise: dict[str, str]
@dataclass
class Ergebnis:
"""Eine Berichtszeile; seite und beschreibung tragen nur Zeilen aus Durchgang 1."""
pfad: Path
aktion: str
text: str
seite: Seitenstand | None = None
beschreibung: Beschreibungsstand | None = None
def seiten_nutzlast(
titel: str, ziel: Ziel, markdown: str, prioritaet: int | None = None
) -> dict[str, object]:
"""name, markdown, chapter_id (wenn ziel.kapitel_id) sonst book_id, priority wenn nicht None."""
daten: dict[str, object] = {"name": titel, "markdown": markdown}
# Bewusst wahr/falsch statt is None: Die Platzhalter-ID 0 geht nie als chapter_id hinaus.
if ziel.kapitel_id:
daten["chapter_id"] = ziel.kapitel_id
else:
daten["book_id"] = ziel.buch_id
if prioritaet is not None:
daten["priority"] = prioritaet
return daten
@@ -0,0 +1,451 @@
"""Kommandozeile: lädt freigegebene Notizen und Beschreibungsdateien nach BookStack, zieht Verweise nach und archiviert.
Ersteller: Pascal Michel
Erstellt: 2026-08-05
Doku: docs/code/hochladen.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
import argparse
import sys
from datetime import datetime
from pathlib import Path
from bookstack.abgleich import Aktion, berechne_pruefsumme, entscheide
from bookstack.api import ApiFehler, BookStackApi
from bookstack.archiv import ARCHIV_ORDNERNAME, archiviere, ist_archiviert
from bookstack.auswahl import (
beschreibungsart,
finde_beschreibungen,
finde_kandidaten,
ist_freigegeben,
titel_von,
)
from bookstack.beschreibung import (
UEBERSICHTSTITEL,
BeschreibungsFehler,
beschreibungs_pruefsumme,
pruefe_beschreibung_html,
pruefe_ziel_felder,
setze_beschreibung,
vorspann,
zu_beschreibungs_html,
)
from bookstack.bilder import Bilduploader
from bookstack.ergebnis import (
ARCHIVIERBAR,
Beschreibungsstand,
Ergebnis,
Seitenstand,
seiten_nutzlast,
)
from bookstack.konfiguration import KonfigurationsFehler, lade_konfiguration
from bookstack.konvertierung import konvertiere, ohne_titelueberschrift
from bookstack.notiz import Notiz, lade_notiz, schreibe_felder
from bookstack.struktur import StrukturFehler, Strukturaufloeser
from bookstack.verweise import Verweisindex, baue_index
from bookstack.verweisrunde import (
baue_endindex,
fuehre_verweisrunde,
haenge_offene_an,
plane_endindex,
)
# Fehler je Datei in Durchgang 1: Sie werden zur FEHLER-Zeile, der Lauf geht weiter.
DATEIFEHLER = (StrukturFehler, ApiFehler, OSError, BeschreibungsFehler)
def _stelle_ausgabe_um() -> None:
"""Windows-Konsolen laufen oft unter cp1252 — „→" bricht den Lauf sonst ab.
Gefunden beim Lauf mit echten Daten: pytest fängt stdout in UTF-8 ab, deshalb
bleibt der Fehler in der Testsuite unsichtbar und schlägt erst beim Anwender zu.
"""
for strom in (sys.stdout, sys.stderr):
umschalten = getattr(strom, "reconfigure", None)
if umschalten is None:
continue
try:
umschalten(encoding="utf-8", errors="replace")
except (ValueError, OSError):
pass
def _argumente(argumente: list[str]) -> argparse.Namespace:
zerleger = argparse.ArgumentParser(
prog="bookstack.hochladen",
description="Lädt freigegebene Vault-Notizen (#review-done) nach BookStack.",
)
zerleger.add_argument("--anwenden", action="store_true", help="tatsächlich schreiben")
zerleger.add_argument(
"--struktur-anlegen", action="store_true", help="fehlende Regale/Bücher/Kapitel anlegen"
)
zerleger.add_argument("--nur", type=Path, help="nur diese eine Datei verarbeiten")
zerleger.add_argument("--baum", action="store_true", help="nur die BookStack-Struktur zeigen")
return zerleger.parse_args(argumente)
def _jetzt() -> str:
return datetime.now().astimezone().isoformat(timespec="seconds")
def _anzeige(pfad: Path, quelle: Path) -> Path:
"""Pfad für den Bericht: relativ zur Quelle, sonst wie er ist."""
try:
return pfad.relative_to(quelle)
except ValueError:
return pfad
def _mit_art(notizen: list[Notiz]) -> list[tuple[Notiz, str]]:
"""Beschreibungsdateien samt Art ('buch' oder 'kapitel'); andere Notizen fallen heraus."""
paare: list[tuple[Notiz, str]] = []
for notiz in notizen:
art = beschreibungsart(notiz.pfad)
if art is not None:
paare.append((notiz, art))
return paare
def verarbeite(
notiz: Notiz,
aufloeser: Strukturaufloeser,
api: BookStackApi,
*,
anwenden: bool,
verweise: Verweisindex,
bilduploader: Bilduploader,
titel: str | None = None,
prioritaet: int | None = None,
) -> Ergebnis:
"""Durchgang 1 für eine Seite: anlegen, aktualisieren oder überspringen.
Das Ergebnis trägt bei jeder dieser Aktionen einen Seitenstand, auch im Probelauf.
Offene Verweise hängt erst haenge_offene_an an, nach der Verweisrunde.
"""
seitentitel = titel_von(notiz) if titel is None else titel
rumpf = ohne_titelueberschrift(notiz.rumpf, seitentitel)
umwandlung = konvertiere(rumpf, verweise=verweise)
ziel = aufloeser.aufloesen(
notiz.frontmatter.get("bookstack_shelf"),
notiz.frontmatter.get("bookstack_book", ""),
notiz.frontmatter.get("bookstack_chapter"),
)
# Bewusst ueber den Pass-1-Markdown: Bild-URLs haengen an der Seiten-ID,
# nicht am Inhalt, und wuerden sonst bei jedem Lauf ein Update ausloesen.
pruefsumme = berechne_pruefsumme(
seitentitel, umwandlung.markdown, ziel.buch_id, ziel.kapitel_id
)
entscheidung = entscheide(notiz, pruefsumme)
def stand(seiten_id: int | None, bilder: dict[str, str] | None) -> Seitenstand:
return Seitenstand(
titel=seitentitel,
rumpf=rumpf,
ziel=ziel,
seiten_id=seiten_id,
prioritaet=prioritaet,
markdown=umwandlung.markdown,
aufgeloeste_verweise=umwandlung.aufgeloeste_verweise,
offene_verweise=umwandlung.offene_verweise,
bildnamen=umwandlung.bildnamen,
bilder=bilder,
)
text = f"{seitentitel} → Buch-ID {ziel.buch_id}"
# is not None: Die Platzhalter-ID 0 eines geplanten Kapitels erscheint im Probelauf.
if ziel.kapitel_id is not None:
text += f", Kapitel-ID {ziel.kapitel_id}"
for angelegt in ziel.angelegt:
text += f"\n Angelegt: {angelegt}"
if entscheidung.aktion is Aktion.UEBERSPRINGEN:
return Ergebnis(
notiz.pfad,
"ÜBERSPRUNGEN",
entscheidung.begruendung,
seite=stand(entscheidung.seiten_id, None),
)
if not anwenden:
# Die Bildhinweise aus Pass 1 sagen nur, dass noch keine URL vorlag; im Probelauf
# steht deshalb „würde hochgeladen“, was wirklich passiert, meldet der Bilduploader.
for name in umwandlung.bildnamen:
text += f"\n Bild würde hochgeladen: {name}"
return Ergebnis(
notiz.pfad,
entscheidung.aktion.name,
text,
seite=stand(entscheidung.seiten_id, None),
)
def nutzlast(markdown: str) -> dict[str, object]:
return seiten_nutzlast(seitentitel, ziel, markdown, prioritaet)
if entscheidung.aktion is Aktion.AKTUALISIEREN:
try:
antwort = api.anfrage(
"PUT", f"/api/pages/{entscheidung.seiten_id}", nutzlast(umwandlung.markdown)
)
except ApiFehler as fehler:
if fehler.status != 404:
raise
# Seite wurde serverseitig gelöscht — neu anlegen statt scheitern.
antwort = api.anfrage("POST", "/api/pages", nutzlast(umwandlung.markdown))
text += "\n Hinweis: Seite war gelöscht, neu angelegt."
else:
antwort = api.anfrage("POST", "/api/pages", nutzlast(umwandlung.markdown))
seiten_id = int(antwort["id"])
# Die ID steht sofort im Frontmatter: Scheitert danach der Bildteil, aktualisiert der nächste
# Lauf diese Seite, statt sie ein zweites Mal anzulegen. Der Hash folgt erst am Ende, damit
# der nächste Lauf eine halb fertige Seite nicht für fertig hält.
schreibe_felder(notiz.pfad, {"bookstack_page_id": str(seiten_id)})
# Die Tabelle geht in den Seitenstand; die Verweisrunde sendet nie ohne sie.
bilder: dict[str, str] = {}
if umwandlung.bildnamen:
bilder, bildhinweise = bilduploader.lade_hoch(
notiz.pfad, seiten_id, umwandlung.bildnamen
)
for hinweis in bildhinweise:
text += f"\n Hinweis: {hinweis}"
if bilder:
zweiter = konvertiere(rumpf, verweise=verweise, bilder=bilder)
api.anfrage("PUT", f"/api/pages/{seiten_id}", nutzlast(zweiter.markdown))
text += f"\n Bilder übertragen: {len(bilder)}"
schreibe_felder(
notiz.pfad,
{
"bookstack_page_id": str(seiten_id),
"bookstack_hash": pruefsumme,
"bookstack_synced": _jetzt(),
},
)
return Ergebnis(notiz.pfad, entscheidung.aktion.name, text, seite=stand(seiten_id, bilder))
def verarbeite_beschreibung(
notiz: Notiz,
art: str,
aufloeser: Strukturaufloeser,
api: BookStackApi,
*,
anwenden: bool,
verweise: Verweisindex,
bilduploader: Bilduploader,
) -> list[Ergebnis]:
"""Durchgang 1 für eine Beschreibungsdatei: erst die Übersichtsseite, dann die Beschreibung.
Fehler in pruefe_ziel_felder und im Seitenteil gehen an fuehre_aus. Fehler im
Beschreibungsteil werden hier zur FEHLER-Zeile; die Seitenzeile bleibt stehen.
"""
pruefe_ziel_felder(art, notiz.frontmatter)
seitenzeile = verarbeite(
notiz,
aufloeser,
api,
anwenden=anwenden,
verweise=verweise,
bilduploader=bilduploader,
titel=UEBERSICHTSTITEL[art],
prioritaet=0,
)
ergebnisse = [seitenzeile]
# verarbeite liefert bei ANLEGEN, AKTUALISIEREN und ÜBERSPRUNGEN immer einen Seitenstand.
assert seitenzeile.seite is not None
ziel = seitenzeile.seite.ziel
ziel_id = ziel.buch_id if art == "buch" else ziel.kapitel_id
kennung = f"Buch-ID {ziel.buch_id}"
if art != "buch":
kennung += f", Kapitel-ID {ziel.kapitel_id}"
try:
if ziel_id is None:
raise BeschreibungsFehler("bookstack_chapter fehlt in der Kapitelbeschreibung")
text = vorspann(notiz.rumpf)
umgewandelt = zu_beschreibungs_html(text, verweise)
pruefe_beschreibung_html(umgewandelt.html)
pruefsumme = beschreibungs_pruefsumme(ziel_id, umgewandelt.html)
stand = Beschreibungsstand(
art=art,
ziel_id=ziel_id,
vorspann=text,
html=umgewandelt.html,
aufgeloeste_verweise=umgewandelt.aufgeloeste_verweise,
)
# unverändert gilt auch im Probelauf; gesendet wird nur mit --anwenden und bei Abweichung.
if notiz.frontmatter.get("bookstack_beschreibung_hash", "").strip() == pruefsumme:
meldung = f"unverändert → {kennung}"
elif not anwenden:
meldung = f"würde gesetzt → {kennung}"
else:
setze_beschreibung(api, art, ziel_id, umgewandelt.html)
schreibe_felder(notiz.pfad, {"bookstack_beschreibung_hash": pruefsumme})
meldung = f"gesetzt → {kennung}"
except (BeschreibungsFehler, ApiFehler, OSError) as fehler:
ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler)))
return ergebnisse
ergebnisse.append(Ergebnis(notiz.pfad, "BESCHREIBUNG", meldung, beschreibung=stand))
return ergebnisse
def _archiviere(gruppen: dict[Path, list[Ergebnis]], quelle: Path, *, anwenden: bool) -> int:
"""Verschiebt je Pfad einmal ins Archiv; Rückgabe: Zahl der ARCHIV-FEHLER.
Archiviert wird ein Pfad mit einer Zeile aus ARCHIVIERBAR und ohne FEHLER-Zeile,
also nur, wenn Seite, Beschreibung und Verweisrunde gelungen sind.
"""
archiviert = 0
fehler_anzahl = 0
for pfad, zeilen in gruppen.items():
aktionen = {ergebnis.aktion for ergebnis in zeilen}
if "FEHLER" in aktionen or aktionen.isdisjoint(ARCHIVIERBAR):
continue
try:
relativ = pfad.relative_to(quelle)
except ValueError:
continue # z. B. --nur mit einem Pfad außerhalb der Quelle — nicht archivieren.
if ist_archiviert(pfad, quelle):
continue
if not anwenden:
print(f" würde archiviert: {relativ}")
continue
try:
neu = archiviere(pfad, quelle)
except OSError as fehler:
# Archivkollision oder gesperrte Datei: Die Notiz bleibt liegen, der Lauf geht weiter.
fehler_anzahl += 1
print(f" ARCHIV-FEHLER {relativ}\n {fehler}")
continue
archiviert += 1
print(f" ARCHIVIERT {relativ} → {neu.relative_to(quelle)}")
if archiviert:
print(f"{archiviert}× ins Archiv verschoben ({ARCHIV_ORDNERNAME})")
return fehler_anzahl
def fuehre_aus(argumente: list[str]) -> int:
_stelle_ausgabe_um()
einstellungen = _argumente(argumente)
try:
konfiguration = lade_konfiguration()
except KonfigurationsFehler as fehler:
print(f"Konfigurationsfehler: {fehler}")
return 1
api = BookStackApi(
konfiguration.basis_url, konfiguration.token_id, konfiguration.token_secret
)
aufloeser = Strukturaufloeser(
api,
anlegen_erlaubt=einstellungen.struktur_anlegen,
nur_planen=not einstellungen.anwenden,
)
bilduploader = Bilduploader(api, konfiguration.vault)
quelle = konfiguration.quelle
anwenden = einstellungen.anwenden
if einstellungen.baum:
try:
print(aufloeser.baum())
except ApiFehler as fehler:
print(f"Fehler: {fehler}")
return 1
return 0
# Auswahl: --nur ordnet die eine Datei über ihren Namen als Beschreibung oder Seite ein.
if einstellungen.nur:
notiz = lade_notiz(einstellungen.nur)
if not ist_freigegeben(notiz):
print(f"{einstellungen.nur} trägt kein #review-done — nichts zu tun.")
return 0
beschreibungen = _mit_art([notiz])
kandidaten = [] if beschreibungen else [notiz]
else:
beschreibungen = _mit_art(finde_beschreibungen(quelle))
kandidaten = finde_kandidaten(quelle)
kopf = "BookStack-Upload" + ("" if anwenden else " (Probelauf)")
print(f"{kopf} — Quelle: {quelle}")
if not beschreibungen and not kandidaten:
print("Keine Notiz mit #review-done gefunden.")
return 0
verweise = baue_index(quelle, konfiguration.basis_url, konfiguration.vault)
# Durchgang 1: erst Beschreibungsdateien (Bücher vor Kapiteln), dann Seiten; der Index bleibt fest.
ergebnisse: list[Ergebnis] = []
for notiz, art in beschreibungen:
try:
ergebnisse.extend(
verarbeite_beschreibung(
notiz,
art,
aufloeser,
api,
anwenden=anwenden,
verweise=verweise,
bilduploader=bilduploader,
)
)
except DATEIFEHLER as fehler:
ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler)))
for notiz in kandidaten:
try:
ergebnisse.append(
verarbeite(
notiz,
aufloeser,
api,
anwenden=anwenden,
verweise=verweise,
bilduploader=bilduploader,
)
)
except DATEIFEHLER as fehler:
ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler)))
# Verweisrunde nur mit --anwenden; der Probelauf plant den Endindex nur und sendet nichts.
if anwenden:
endindex = baue_endindex(
verweise,
ergebnisse,
quelle=quelle,
vault=konfiguration.vault,
basis_url=konfiguration.basis_url,
)
ergebnisse += fuehre_verweisrunde(ergebnisse, endindex, api, bilduploader)
else:
endindex = plane_endindex(verweise, ergebnisse, quelle=quelle, vault=konfiguration.vault)
aufgeloest = haenge_offene_an(ergebnisse, endindex)
# Bericht nach Pfad gruppiert; Zeilen der Verweisrunde stehen unter den Zeilen ihrer Datei.
gruppen: dict[Path, list[Ergebnis]] = {}
for ergebnis in ergebnisse:
gruppen.setdefault(ergebnis.pfad, []).append(ergebnis)
for pfad, zeilen in gruppen.items():
anzeige = _anzeige(pfad, quelle)
for ergebnis in zeilen:
print(f" {ergebnis.aktion:<14} {anzeige}\n {ergebnis.text}")
zaehler: dict[str, int] = {}
for ergebnis in ergebnisse:
zaehler[ergebnis.aktion] = zaehler.get(ergebnis.aktion, 0) + 1
print("\n" + ", ".join(f"{anzahl}× {name}" for name, anzahl in sorted(zaehler.items())))
if not anwenden and aufgeloest:
print(f"{aufgeloest}× Verweis wird in der Verweisrunde aufgelöst")
archiv_fehler = _archiviere(gruppen, quelle, anwenden=anwenden)
if not anwenden:
print("Probelauf — nichts geschrieben. Mit --anwenden ausführen.")
return 1 if zaehler.get("FEHLER", 0) or archiv_fehler else 0
def main() -> int:
return fuehre_aus(sys.argv[1:])
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,54 @@
"""Liest die BookStack-Zugangsdaten aus Umgebungsvariablen."""
from __future__ import annotations
import os
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
STANDARD_QUELLE = Path(r"D:\Claude\Bookstack")
STANDARD_VAULT = Path(r"D:\Claude")
PFLICHTVARIABLEN = ("BOOKSTACK_URL", "BOOKSTACK_TOKEN_ID", "BOOKSTACK_TOKEN_SECRET")
class KonfigurationsFehler(Exception):
"""Zugangsdaten fehlen oder sind unbrauchbar."""
@dataclass(frozen=True)
class Konfiguration:
basis_url: str
token_id: str
token_secret: str = field(repr=False)
quelle: Path
vault: Path
def lade_konfiguration(umgebung: Mapping[str, str] | None = None) -> Konfiguration:
"""Baut die Konfiguration aus der Umgebung; meldet alle Lücken auf einmal."""
werte = os.environ if umgebung is None else umgebung
fehlend = [name for name in PFLICHTVARIABLEN if not werte.get(name, "").strip()]
if fehlend:
raise KonfigurationsFehler(
"Diese Umgebungsvariablen fehlen oder sind leer: "
+ ", ".join(fehlend)
+ ". Siehe README.md, Abschnitt „Einrichtung“."
)
basis_url = werte["BOOKSTACK_URL"].strip().rstrip("/")
if not basis_url.startswith(("http://", "https://")):
raise KonfigurationsFehler(
f"BOOKSTACK_URL muss mit http:// oder https:// beginnen, ist aber „{basis_url}“."
)
quelle = werte.get("BOOKSTACK_QUELLE", "").strip()
vault = werte.get("BOOKSTACK_VAULT", "").strip()
return Konfiguration(
basis_url=basis_url,
token_id=werte["BOOKSTACK_TOKEN_ID"].strip(),
token_secret=werte["BOOKSTACK_TOKEN_SECRET"].strip(),
quelle=Path(quelle) if quelle else STANDARD_QUELLE,
vault=Path(vault) if vault else STANDARD_VAULT,
)
@@ -0,0 +1,143 @@
"""Übersetzt Obsidian-Eigenheiten in Markdown, das BookStack versteht.
Ersteller: Pascal Michel
Erstellt: 2026-08-05
Doku: docs/code/konvertierung.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
import re
from collections.abc import Mapping
from dataclasses import dataclass, field
from bookstack.verweise import Verweisindex
EINBETTUNG = re.compile(r"!\[\[([^\]|]+?)(?:\|[^\]]*)?\]\]")
WIKILINK = re.compile(r"\[\[([^\]|]+?)(?:\|([^\]]*))?\]\]")
REVIEW_TAG = re.compile(r"(?<![\w#-])#review(?:-done)?(?![\w-])[ \t]*")
CALLOUT = re.compile(r"^>\s*\[!(\w+)\][ \t]*(.*)$", re.MULTILINE)
ZAUN = re.compile(r"(```.*?```|~~~.*?~~~)", re.DOTALL)
TITELZEILE = re.compile(r"^#\s+(.+?)[ \t]*$", re.MULTILINE)
MAX_TITEL = 255
@dataclass
class Umwandlung:
markdown: str
hinweise: list[str] = field(default_factory=list)
bildnamen: list[str] = field(default_factory=list)
offene_verweise: list[str] = field(default_factory=list) # Einträge: ziel aus zerlege_wikilink
aufgeloeste_verweise: dict[str, str] = field(default_factory=dict) # ziel → URL
@dataclass
class Sammler:
verweise: Verweisindex
bilder: Mapping[str, str]
hinweise: list[str] = field(default_factory=list)
bildnamen: list[str] = field(default_factory=list)
offene_verweise: list[str] = field(default_factory=list)
aufgeloeste_verweise: dict[str, str] = field(default_factory=dict)
def _ohne_tabellen_escape(text: str) -> str:
# In Markdown-Tabellen steht [[Ziel\|Alias]]; der Backslash gehört nicht zum Ziel.
text = text.strip()
if text.endswith("\\"):
text = text[:-1]
return text.strip()
def zerlege_wikilink(ziel_roh: str, alias: str | None) -> tuple[str, str]:
"""(ziel, anzeigetext).
ziel: ziel_roh ohne Anker, ohne abschließenden Backslash, gestrippt, sonst wie geschrieben.
anzeigetext: alias gestrippt, sonst letztes Segment von ziel."""
ziel = _ohne_tabellen_escape(ziel_roh.split("#", 1)[0])
anzeigetext = (alias or "").strip() or ziel.split("/")[-1]
return ziel, anzeigetext
def ankertext(ziel_roh: str, alias: str | None) -> str | None:
"""Anzeigetext eines reinen Ankerlinks wie [[#Abschnitt]] oder [[#Abschnitt|Text]], sonst None.
Alias gestrippt, sonst der Anker ohne '#'."""
ziel = _ohne_tabellen_escape(ziel_roh)
if not ziel.startswith("#"):
return None
return (alias or "").strip() or ziel.lstrip("#").strip()
def _ersetze_ausserhalb_von_code(text: str, sammler: "Sammler") -> str:
teile = ZAUN.split(text)
for stelle, teil in enumerate(teile):
if stelle % 2 == 1: # ungerade Indizes sind die Codeblöcke
continue
teile[stelle] = _ersetze(teil, sammler)
return "".join(teile)
def _ersetze(text: str, sammler: Sammler) -> str:
def bild(treffer: re.Match[str]) -> str:
name = _ohne_tabellen_escape(treffer.group(1))
if name not in sammler.bildnamen:
sammler.bildnamen.append(name)
url = sammler.bilder.get(name)
if url:
return f"![{name}]({url})"
sammler.hinweise.append(f"Bild nicht übertragen: {name}")
return f"*(Bild „{name}“ wurde nicht übertragen)*"
def link(treffer: re.Match[str]) -> str:
anker = ankertext(treffer.group(1), treffer.group(2))
if anker is not None:
# Ein Anker auf die eigene Seite ist kein Verweis: Klartext, nichts in offene_verweise.
return anker
ziel, anzeigetext = zerlege_wikilink(treffer.group(1), treffer.group(2))
url = sammler.verweise.aufloesen(ziel)
if url:
sammler.aufgeloeste_verweise[ziel] = url
return f"[{anzeigetext}]({url})"
if ziel not in sammler.offene_verweise:
sammler.offene_verweise.append(ziel)
return anzeigetext
def callout(treffer: re.Match[str]) -> str:
titel = treffer.group(2).strip() or treffer.group(1).capitalize()
return f"> **{titel}**"
text = EINBETTUNG.sub(bild, text)
text = WIKILINK.sub(link, text)
text = CALLOUT.sub(callout, text)
return REVIEW_TAG.sub("", text)
def ohne_titelueberschrift(rumpf: str, titel: str) -> str:
"""Entfernt die erste H1, wenn sie den Seitentitel wiederholt.
BookStack stellt den Seitennamen ohnehin als Überschrift dar. Bliebe die H1
im Text stehen, stünde die Überschrift zweimal auf der Seite.
Weicht die H1 vom Titel ab, bleibt sie unangetastet.
"""
treffer = TITELZEILE.search(rumpf)
if treffer is None or treffer.group(1).strip()[:MAX_TITEL] != titel:
return rumpf
return rumpf[: treffer.start()] + rumpf[treffer.end() :].lstrip("\n")
def konvertiere(
rumpf: str,
*,
verweise: Mapping[str, str] | None = None, # Verweisindex oder schlichtes dict
bilder: Mapping[str, str] | None = None,
) -> Umwandlung:
index = verweise if isinstance(verweise, Verweisindex) else Verweisindex(verweise)
sammler = Sammler(verweise=index, bilder=bilder or {})
markdown = _ersetze_ausserhalb_von_code(rumpf, sammler)
markdown = re.sub(r"\n{3,}", "\n\n", markdown).strip() + "\n"
return Umwandlung(
markdown=markdown,
hinweise=sammler.hinweise,
bildnamen=sammler.bildnamen,
offene_verweise=sammler.offene_verweise,
aufgeloeste_verweise=sammler.aufgeloeste_verweise,
)
@@ -0,0 +1,98 @@
"""Liest YAML-Frontmatter und schreibt einzelne Felder zurück, ohne den Rest anzufassen."""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
TRENNER = "---"
SKALARE_ZEILE = re.compile(r"^([A-Za-z_][A-Za-z0-9_-]*):\s*(.*)$")
@dataclass(frozen=True)
class Notiz:
pfad: Path
frontmatter: dict[str, str]
rumpf: str
roh: str
def _zerlege(roh: str) -> tuple[list[str], str]:
"""Liefert (Frontmatter-Zeilen ohne Trenner, Rumpf). Ohne Frontmatter: ([], roh)."""
if not roh.startswith(TRENNER + "\n"):
return [], roh
rest = roh[len(TRENNER) + 1 :]
ende = rest.find("\n" + TRENNER + "\n")
if ende == -1:
return [], roh
kopf = rest[:ende].split("\n")
rumpf = rest[ende + len(TRENNER) + 2 :]
return kopf, rumpf.lstrip("\n")
def _entpacke(wert: str) -> str:
wert = wert.strip()
if len(wert) >= 2 and wert[0] == wert[-1] and wert[0] in "\"'":
return wert[1:-1]
return wert
def _verpacke(wert: str) -> str:
"""Quotet nur, wo YAML es wirklich braucht.
Ein Doppelpunkt allein reicht nicht: „sha256:abc" und „2026-08-05T12:00:00+02:00"
sind gültige einfache Skalare. Erst „Doppelpunkt plus Leerzeichen" beendet einen
Schlüssel. Würde hier jeder Doppelpunkt quoten, stünde in jeder Notiz
bookstack_hash: "sha256:…" statt bookstack_hash: sha256:…
"""
if not wert:
return '""'
unsicher = (
": " in wert
or wert.endswith(":")
or " #" in wert
or "\n" in wert
or wert != wert.strip()
or wert[0] in "\"'[]{}>|*&!%@`#"
)
if unsicher:
return '"' + wert.replace('"', '\\"') + '"'
return wert
def lade_notiz(pfad: Path) -> Notiz:
roh = pfad.read_text(encoding="utf-8")
kopf, rumpf = _zerlege(roh)
felder: dict[str, str] = {}
for zeile in kopf:
treffer = SKALARE_ZEILE.match(zeile)
if treffer and not treffer.group(2).startswith("["):
felder[treffer.group(1)] = _entpacke(treffer.group(2))
return Notiz(pfad=pfad, frontmatter=felder, rumpf=rumpf, roh=roh)
def setze_felder(roh: str, werte: dict[str, str]) -> str:
"""Setzt oder ersetzt Frontmatter-Felder und lässt alles andere zeichengleich."""
kopf, rumpf = _zerlege(roh)
if not kopf and not roh.startswith(TRENNER + "\n"):
neue = [f"{name}: {_verpacke(wert)}" for name, wert in werte.items()]
return TRENNER + "\n" + "\n".join(neue) + "\n" + TRENNER + "\n" + roh
offen = dict(werte)
ergebnis: list[str] = []
for zeile in kopf:
treffer = SKALARE_ZEILE.match(zeile)
name = treffer.group(1) if treffer else None
if name in offen:
ergebnis.append(f"{name}: {_verpacke(offen.pop(name))}")
else:
ergebnis.append(zeile)
ergebnis.extend(f"{name}: {_verpacke(wert)}" for name, wert in offen.items())
return TRENNER + "\n" + "\n".join(ergebnis) + "\n" + TRENNER + "\n\n" + rumpf
def schreibe_felder(pfad: Path, werte: dict[str, str]) -> None:
roh = pfad.read_text(encoding="utf-8")
pfad.write_text(setze_felder(roh, werte), encoding="utf-8", newline="\n")
@@ -0,0 +1,184 @@
"""Löst Regal-, Buch- und Kapitelnamen in IDs auf und legt Fehlendes auf Wunsch an.
Ersteller: Pascal Michel
Erstellt: 2026-08-05
Doku: docs/code/struktur.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
from dataclasses import dataclass, field
from bookstack.api import BookStackApi
class StrukturFehler(Exception):
"""Ein benanntes Regal, Buch oder Kapitel existiert nicht (und darf nicht angelegt werden)."""
@dataclass
class Ziel:
buch_id: int # 0: Buch ist im Probelauf nur geplant
kapitel_id: int | None = None # 0: Kapitel nur geplant, None: kein Kapitel
angelegt: list[str] = field(default_factory=list)
def _passt(eintrag: dict, name: str) -> bool:
return str(eintrag.get("name", "")).strip().casefold() == name.strip().casefold()
def _geplant_schluessel(sammlung: str, *namen: str) -> tuple[str, ...]:
"""Schlüssel für das Probelauf-Gedächtnis; Namen vergleicht er wie _passt."""
return (sammlung, *(name.strip().casefold() for name in namen))
class Strukturaufloeser:
def __init__(
self,
api: BookStackApi,
*,
anlegen_erlaubt: bool = False,
nur_planen: bool = False,
) -> None:
self._api = api
self._anlegen_erlaubt = anlegen_erlaubt
# Im Probelauf darf nichts entstehen. Fehlendes wird nur benannt.
self._nur_planen = nur_planen
self._speicher: dict[str, list[dict]] = {}
# Probelauf: schon als „wird angelegt“ gemeldete Regale, Bücher und Kapitel.
self._geplant: set[tuple[str, ...]] = set()
def _sammlung(self, name: str) -> list[dict]:
if name not in self._speicher:
self._speicher[name] = self._api.liste(f"/api/{name}")
return self._speicher[name]
def _verwerfe(self, name: str) -> None:
self._speicher.pop(name, None)
def _finde(self, sammlung: str, name: str, **filter_felder) -> dict | None:
for eintrag in self._sammlung(sammlung):
if _passt(eintrag, name) and all(
eintrag.get(feld) == wert for feld, wert in filter_felder.items()
):
return eintrag
return None
def _fehlt(self, art: str, name: str) -> StrukturFehler:
return StrukturFehler(
f"{art} „{name}“ existiert in BookStack nicht. "
"Namen im Frontmatter prüfen oder mit --struktur-anlegen erneut starten."
)
def _vermerke(self, art: str, name: str) -> str:
"""Beschreibt eine Neuanlage — im Probelauf als Absicht, sonst als Tatsache."""
zusatz = " (wird angelegt)" if self._nur_planen else ""
return f'{art} "{name.strip()}"{zusatz}'
def _plane(
self, schluessel: tuple[str, ...], art: str, name: str, angelegt: list[str]
) -> None:
"""Probelauf: meldet eine geplante Neuanlage nur beim ersten Mal."""
if schluessel in self._geplant:
return
self._geplant.add(schluessel)
angelegt.append(self._vermerke(art, name))
def _haenge_an_regal(self, regal_name: str, buch_id: int, angelegt: list[str]) -> None:
regal = self._finde("shelves", regal_name)
if regal is None:
if not self._anlegen_erlaubt:
raise self._fehlt("Regal", regal_name)
if self._nur_planen:
self._plane(
_geplant_schluessel("shelves", regal_name), "Regal", regal_name, angelegt
)
return
angelegt.append(self._vermerke("Regal", regal_name))
self._api.anfrage(
"POST", "/api/shelves", {"name": regal_name.strip(), "books": [buch_id]}
)
self._verwerfe("shelves")
return
vollstaendig = self._api.anfrage("GET", f"/api/shelves/{regal['id']}")
# Lesen liefert Buchobjekte, Schreiben erwartet reine IDs. Beide Formen dulden.
vorhandene = [
int(eintrag["id"] if isinstance(eintrag, dict) else eintrag)
for eintrag in vollstaendig.get("books", [])
]
if buch_id in vorhandene or self._nur_planen:
return
self._api.anfrage(
"PUT", f"/api/shelves/{regal['id']}", {"books": vorhandene + [buch_id]}
)
def aufloesen(self, regal: str | None, buch: str, kapitel: str | None) -> Ziel:
if not buch or not buch.strip():
raise StrukturFehler("Feld „bookstack_book“ fehlt oder ist leer.")
angelegt: list[str] = []
gefunden = self._finde("books", buch)
if gefunden is None:
if not self._anlegen_erlaubt:
raise self._fehlt("Buch", buch)
if self._nur_planen:
self._plane(_geplant_schluessel("books", buch), "Buch", buch, angelegt)
gefunden = {"id": 0}
else:
angelegt.append(self._vermerke("Buch", buch))
gefunden = self._api.anfrage("POST", "/api/books", {"name": buch.strip()})
self._verwerfe("books")
buch_id = int(gefunden["id"])
if regal and regal.strip():
self._haenge_an_regal(regal, buch_id, angelegt)
kapitel_id: int | None = None
if kapitel and kapitel.strip():
treffer = self._finde("chapters", kapitel, book_id=buch_id)
if treffer is None:
if not self._anlegen_erlaubt:
raise self._fehlt("Kapitel", kapitel)
if self._nur_planen:
self._plane(
_geplant_schluessel("chapters", buch, kapitel), "Kapitel", kapitel, angelegt
)
treffer = {"id": 0}
else:
angelegt.append(self._vermerke("Kapitel", kapitel))
treffer = self._api.anfrage(
"POST", "/api/chapters", {"book_id": buch_id, "name": kapitel.strip()}
)
self._verwerfe("chapters")
kapitel_id = int(treffer["id"])
return Ziel(buch_id=buch_id, kapitel_id=kapitel_id, angelegt=angelegt)
def baum(self) -> str:
"""Menschenlesbarer Überblick über die vorhandene Struktur."""
buecher = {int(eintrag["id"]): eintrag for eintrag in self._sammlung("books")}
kapitel: dict[int, list[str]] = {}
for eintrag in self._sammlung("chapters"):
kapitel.setdefault(int(eintrag.get("book_id", 0)), []).append(str(eintrag["name"]))
zeilen: list[str] = []
zugeordnet: set[int] = set()
for regal in self._sammlung("shelves"):
zeilen.append(f"Regal: {regal['name']}")
vollstaendig = self._api.anfrage("GET", f"/api/shelves/{regal['id']}")
for eintrag in vollstaendig.get("books", []):
buch_id = int(eintrag["id"])
zugeordnet.add(buch_id)
zeilen.append(f" Buch: {buecher.get(buch_id, eintrag)['name']}")
zeilen.extend(f" Kapitel: {name}" for name in kapitel.get(buch_id, []))
frei = [wert for schluessel, wert in buecher.items() if schluessel not in zugeordnet]
if frei:
zeilen.append("Ohne Regal:")
for eintrag in frei:
zeilen.append(f" Buch: {eintrag['name']}")
zeilen.extend(
f" Kapitel: {name}" for name in kapitel.get(int(eintrag["id"]), [])
)
return "\n".join(zeilen)
@@ -0,0 +1,171 @@
"""Baut den Verweisindex hochgeladener Notizen und löst Wikilink-Ziele darüber auf.
Ersteller: Pascal Michel
Erstellt: 2026-08-05
Doku: docs/code/verweise.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
from collections.abc import Iterable, Iterator, Mapping
from pathlib import Path
from bookstack.archiv import ARCHIV_ORDNERNAME
from bookstack.auswahl import beschreibungsart
from bookstack.notiz import lade_notiz
def permalink(basis_url: str, seiten_id: int) -> str:
"""Stabiler Seitenlink.
BookStack löst `/link/{id}` auf die Seite auf. Das überlebt Umbenennungen,
anders als der Weg über Buch- und Seiten-Slug — und `GET /api/pages/{id}`
liefert `book_slug` ohnehin nicht mit.
"""
return f"{basis_url.rstrip('/')}/link/{seiten_id}"
def normalisiere_ziel(ziel: str) -> str:
"""Anker ab '#' weg, Tabellen-Escape (abschließender Backslash) weg, '\\' → '/',
führende '/' weg, abschließendes '.md' weg (ohne Rücksicht auf Groß/klein), casefold, strip."""
text = ziel.split("#", 1)[0].strip()
if text.endswith("\\"):
text = text[:-1]
text = text.replace("\\", "/").strip().lstrip("/")
if text.casefold().endswith(".md"):
text = text[:-3]
return text.casefold().strip()
def _stamm(schluessel: str) -> str:
return schluessel.rsplit("/", 1)[-1]
def _relativ(pfad: Path, basis: Path) -> Path | None:
try:
return pfad.resolve().relative_to(basis.resolve())
except ValueError:
return None
def schluessel_fuer(pfad: Path, quelle: Path, vault: Path | None = None) -> list[str]:
"""Indexschlüssel einer Notiz.
Basis ist vault, wenn gesetzt und pfad darunter liegt, sonst quelle (Vergleich nach resolve()).
Erster Schlüssel: pfad relativ zur Basis, normalisiert.
Liegt pfad unter quelle/…/ARCHIV_ORDNERNAME/…, zweiter Schlüssel für denselben Pfad
ohne das erste Vorkommen von ARCHIV_ORDNERNAME. Liegt pfad unter keiner Basis: []."""
relativ = _relativ(pfad, vault) if vault is not None else None
if relativ is None:
relativ = _relativ(pfad, quelle)
if relativ is None:
return []
schluessel = [normalisiere_ziel(relativ.as_posix())]
in_quelle = _relativ(pfad, quelle)
if in_quelle is not None and ARCHIV_ORDNERNAME in in_quelle.parts[:-1]:
# Beide Pfade enden auf derselben Datei; gezählt wird von hinten, damit das
# erste Vorkommen unterhalb der Quelle fällt, nicht ein gleichnamiger Ordner darüber.
stelle = (
len(relativ.parts) - len(in_quelle.parts) + in_quelle.parts.index(ARCHIV_ORDNERNAME)
)
if stelle >= 0:
teile = relativ.parts[:stelle] + relativ.parts[stelle + 1 :]
schluessel.append(normalisiere_ziel("/".join(teile)))
return schluessel
class Verweisindex(Mapping[str, str]):
"""Normalisiertes Wikilink-Ziel → Permalink, mit eindeutigem Rückfall über Pfadende oder Dateinamen."""
def __init__(
self, eintraege: Mapping[str, str] | None = None, aliase: Iterable[str] = ()
) -> None:
self._eintraege: dict[str, str] = {}
for schluessel, url in (eintraege or {}).items():
self._eintraege[normalisiere_ziel(schluessel)] = url
# Archiv-Zusatzschlüssel: Nur sie darf ein echter Schlüssel einer anderen Notiz verdrängen.
self._aliase: frozenset[str] = frozenset(
eintrag for eintrag in map(normalisiere_ziel, aliase) if eintrag in self._eintraege
)
self._staemme: dict[str, set[str]] = {}
for schluessel, url in self._eintraege.items():
self._staemme.setdefault(_stamm(schluessel), set()).add(url)
def __getitem__(self, schluessel: str) -> str:
return self._eintraege[schluessel]
def __iter__(self) -> Iterator[str]:
return iter(self._eintraege)
def __len__(self) -> int:
return len(self._eintraege)
def __repr__(self) -> str:
return f"Verweisindex({self._eintraege!r})"
def aufloesen(self, ziel: str) -> str | None:
schluessel = normalisiere_ziel(ziel)
if not schluessel:
return None
if schluessel in self._eintraege:
return self._eintraege[schluessel]
if "/" in schluessel:
# Ein Ziel mit Pfad trifft nur ganze Pfadenden, nie den Dateinamen allein: Sonst landete
# [[Y/_Kapitelbeschreibung]] still auf dem einzigen Kapitel, das der Index kennt.
endung = "/" + schluessel
urls = {url for eintrag, url in self._eintraege.items() if eintrag.endswith(endung)}
else:
# Nur eindeutig: _kapitelbeschreibung gibt es je Kapitel einmal, der Stamm allein darf dann nie treffen.
urls = self._staemme.get(schluessel, set())
if len(urls) == 1:
return next(iter(urls))
return None
def ergaenzt(self, weitere: Mapping[str, str], aliase: Iterable[str] = ()) -> Verweisindex:
"""Neuer Index; bei gleichem Schlüssel gewinnt weitere.
Ausnahme: Ein Schlüssel aus aliase verdrängt keinen echten Schlüssel, nur einen Alias."""
neue_aliase = {normalisiere_ziel(schluessel) for schluessel in aliase}
zusammen = dict(self._eintraege)
zusammen_aliase = set(self._aliase)
for schluessel, url in weitere.items():
eintrag = normalisiere_ziel(schluessel)
if eintrag not in neue_aliase:
zusammen[eintrag] = url
zusammen_aliase.discard(eintrag)
elif eintrag not in zusammen or eintrag in zusammen_aliase:
zusammen[eintrag] = url
zusammen_aliase.add(eintrag)
return Verweisindex(zusammen, zusammen_aliase)
def trage_ein(eintraege: dict[str, str], aliase: set[str], schluessel: list[str], url: str) -> None:
"""Schlüssel einer Notiz aus schluessel_fuer: der erste echt, weitere als Alias.
Ein echter Schlüssel überschreibt einen Alias; ein Alias belegt nur, was kein echter Schlüssel belegt."""
for stelle, eintrag in enumerate(schluessel):
if stelle == 0:
eintraege[eintrag] = url
aliase.discard(eintrag)
elif eintrag not in eintraege or eintrag in aliase:
eintraege[eintrag] = url
aliase.add(eintrag)
def baue_index(quelle: Path, basis_url: str, vault: Path | None = None) -> Verweisindex:
"""rglob('*.md') sortiert. Überspringt Namen ab '.' oder '_', außer beschreibungsart(pfad)
ist nicht None. Nimmt nur ganzzahlige bookstack_page_id. Jeder Schlüssel aus
schluessel_fuer → permalink, eingetragen über trage_ein."""
if not quelle.is_dir():
return Verweisindex()
eintraege: dict[str, str] = {}
aliase: set[str] = set()
for pfad in sorted(quelle.rglob("*.md")):
if pfad.name.startswith((".", "_")) and beschreibungsart(pfad) is None:
continue
roh_id = lade_notiz(pfad).frontmatter.get("bookstack_page_id", "").strip()
try:
seiten_id = int(roh_id)
except ValueError:
continue
# Nicht die Sortierung entscheidet: Eine archivierte Notiz verdrängt nie die neue am alten Pfad.
schluessel = schluessel_fuer(pfad, quelle, vault)
trage_ein(eintraege, aliase, schluessel, permalink(basis_url, seiten_id))
return Verweisindex(eintraege, aliase)
@@ -0,0 +1,206 @@
"""Zieht nach Durchgang 1 die Verweise auf Seiten desselben Laufs nach und meldet, was offen bleibt.
Ersteller: Pascal Michel
Erstellt: 2026-09-30
Doku: docs/code/verweisrunde.md · Version und Änderungen: CHANGELOG.md
"""
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from bookstack.abgleich import berechne_pruefsumme
from bookstack.api import ApiFehler, BookStackApi
from bookstack.beschreibung import (
BeschreibungsFehler,
beschreibungs_pruefsumme,
pruefe_beschreibung_html,
setze_beschreibung,
zu_beschreibungs_html,
)
from bookstack.bilder import Bilduploader
from bookstack.ergebnis import (
ARCHIVIERBAR,
Beschreibungsstand,
Ergebnis,
Seitenstand,
seiten_nutzlast,
)
from bookstack.konvertierung import konvertiere
from bookstack.notiz import schreibe_felder
from bookstack.verweise import Verweisindex, permalink, schluessel_fuer, trage_ein
PLATZHALTER_PRAEFIX = "geplant:"
def _jetzt() -> str:
return datetime.now().astimezone().isoformat(timespec="seconds")
def _neu_oder_geaendert(vorher: dict[str, str], nachher: dict[str, str]) -> int:
"""Zahl der Verweisziele, deren URL neu dazukam oder sich geändert hat."""
return sum(1 for ziel, url in nachher.items() if vorher.get(ziel) != url)
def baue_endindex(
verweise: Verweisindex, ergebnisse: list[Ergebnis], *, quelle: Path, vault: Path, basis_url: str
) -> Verweisindex:
"""verweise.ergaenzt(...) um jede Zeile mit Aktion aus ARCHIVIERBAR und seite.seiten_id
nicht None: alle schluessel_fuer(pfad) → permalink(basis_url, seiten_id)."""
weitere: dict[str, str] = {}
aliase: set[str] = set()
for ergebnis in ergebnisse:
seite = ergebnis.seite
if ergebnis.aktion not in ARCHIVIERBAR or seite is None or seite.seiten_id is None:
continue
# Über trage_ein, nicht per Zuweisung: Sonst entschiede die Reihenfolge der Zeilen, ob der
# Alias einer archivierten Notiz die neue Notiz am alten Pfad verdrängt.
trage_ein(
weitere,
aliase,
schluessel_fuer(ergebnis.pfad, quelle, vault),
permalink(basis_url, seite.seiten_id),
)
return verweise.ergaenzt(weitere, aliase)
def plane_endindex(
verweise: Verweisindex, ergebnisse: list[Ergebnis], *, quelle: Path, vault: Path
) -> Verweisindex:
"""Probelauf: jede Zeile mit Aktion 'ANLEGEN' und seite → PLATZHALTER_PRAEFIX + pfad.as_posix()."""
weitere: dict[str, str] = {}
aliase: set[str] = set()
for ergebnis in ergebnisse:
if ergebnis.aktion != "ANLEGEN" or ergebnis.seite is None:
continue
trage_ein(
weitere,
aliase,
schluessel_fuer(ergebnis.pfad, quelle, vault),
PLATZHALTER_PRAEFIX + ergebnis.pfad.as_posix(),
)
return verweise.ergaenzt(weitere, aliase)
def _bildtabelle(
seite: Seitenstand, seiten_id: int, bilduploader: Bilduploader
) -> tuple[dict[str, str], list[str]]:
"""(Tabelle für den PUT, fehlende Bildnamen). Lädt nie hoch, liest höchstens die Galerie."""
if seite.bilder is not None:
# Tabelle aus Durchgang 1; deckt sie nicht alle Bildnamen ab, steht das unter fehlend.
tabelle = seite.bilder
elif not seite.bildnamen:
return {}, []
else:
# Durchgang 1 hat nichts gesendet (ÜBERSPRUNGEN): Die Tabelle kommt lesend aus der Galerie.
vorhanden = bilduploader.vorhandene(seiten_id)
tabelle = {name: vorhanden[name] for name in seite.bildnamen if vorhanden.get(name)}
fehlend = [name for name in dict.fromkeys(seite.bildnamen) if name not in tabelle]
return tabelle, fehlend
def _ziehe_seite_nach(
pfad: Path,
seite: Seitenstand,
seiten_id: int,
endindex: Verweisindex,
api: BookStackApi,
bilduploader: Bilduploader,
) -> Ergebnis | None:
neu = konvertiere(seite.rumpf, verweise=endindex)
if neu.markdown == seite.markdown:
return None
tabelle, fehlend = _bildtabelle(seite, seiten_id, bilduploader)
# Nie ein PUT mit Platzhaltertext: Fehlt ein Bildname in der Tabelle, gibt es nur die Zeile.
if fehlend:
namen = ", ".join(fehlend)
return Ergebnis(
pfad,
"FEHLER",
f"Verweise nicht nachgezogen: Bild fehlt in der Galerie der Seite ({namen}). "
"Der nächste Lauf aktualisiert die Seite.",
)
markdown = neu.markdown
if seite.bildnamen:
markdown = konvertiere(seite.rumpf, verweise=endindex, bilder=tabelle).markdown
api.anfrage(
"PUT",
f"/api/pages/{seiten_id}",
seiten_nutzlast(seite.titel, seite.ziel, markdown, seite.prioritaet),
)
# Gespeichert wird die Summe des bildfreien Markdowns mit den Verweisen dieses PUT.
pruefsumme = berechne_pruefsumme(
seite.titel, neu.markdown, seite.ziel.buch_id, seite.ziel.kapitel_id
)
schreibe_felder(pfad, {"bookstack_hash": pruefsumme, "bookstack_synced": _jetzt()})
anzahl = _neu_oder_geaendert(seite.aufgeloeste_verweise, neu.aufgeloeste_verweise)
return Ergebnis(pfad, "VERLINKT", f"{anzahl} Verweise nachgezogen")
def _ziehe_beschreibung_nach(
pfad: Path, stand: Beschreibungsstand, endindex: Verweisindex, api: BookStackApi
) -> Ergebnis | None:
neu = zu_beschreibungs_html(stand.vorspann, endindex)
if neu.html == stand.html:
return None
pruefe_beschreibung_html(neu.html)
setze_beschreibung(api, stand.art, stand.ziel_id, neu.html)
schreibe_felder(
pfad, {"bookstack_beschreibung_hash": beschreibungs_pruefsumme(stand.ziel_id, neu.html)}
)
anzahl = _neu_oder_geaendert(stand.aufgeloeste_verweise, neu.aufgeloeste_verweise)
return Ergebnis(pfad, "VERLINKT", f"{anzahl} Verweise in der Beschreibung nachgezogen")
def fuehre_verweisrunde(
ergebnisse: list[Ergebnis],
endindex: Verweisindex,
api: BookStackApi,
bilduploader: Bilduploader,
) -> list[Ergebnis]:
"""Zweiter Durchgang, nur mit --anwenden. Liefert neue VERLINKT- und FEHLER-Zeilen."""
neue: list[Ergebnis] = []
for ergebnis in ergebnisse:
seite = ergebnis.seite
beschreibung = ergebnis.beschreibung
try:
if (
seite is not None
and seite.seiten_id is not None
and ergebnis.aktion in ARCHIVIERBAR
):
zeile = _ziehe_seite_nach(
ergebnis.pfad, seite, seite.seiten_id, endindex, api, bilduploader
)
elif beschreibung is not None and ergebnis.aktion == "BESCHREIBUNG":
zeile = _ziehe_beschreibung_nach(ergebnis.pfad, beschreibung, endindex, api)
else:
continue
except (ApiFehler, OSError, BeschreibungsFehler) as fehler:
zeile = Ergebnis(ergebnis.pfad, "FEHLER", str(fehler))
if zeile is not None:
neue.append(zeile)
return neue
def haenge_offene_an(ergebnisse: list[Ergebnis], endindex: Verweisindex) -> int:
"""Hängt an jede Zeile mit seite „Verweis nicht auflösbar“ für zwei Gruppen von Zielen an,
soweit endindex sie nicht auflöst: die offenen Verweise aus Durchgang 1 (seite.offene_verweise)
und die Ziele aus seite.aufgeloeste_verweise. Rückgabe: Zahl der offenen Verweise, die erst
der Endindex auflöst; die Ziele aus aufgeloeste_verweise werden gemeldet, nicht gezählt."""
aufgeloest = 0
for ergebnis in ergebnisse:
if ergebnis.seite is None:
continue
for ziel in ergebnis.seite.offene_verweise:
if endindex.aufloesen(ziel) is None:
ergebnis.text += f"\n Verweis nicht auflösbar: {ziel}"
else:
aufgeloest += 1
# Ein Stamm, der in Durchgang 1 eindeutig war, kann im Endindex mehrdeutig werden;
# der PUT der Runde entfernt den Link dann, und das darf nicht still passieren.
for ziel in ergebnis.seite.aufgeloeste_verweise:
if endindex.aufloesen(ziel) is None:
ergebnis.text += f"\n Verweis nicht auflösbar: {ziel}"
return aufgeloest
@@ -0,0 +1,62 @@
---
name: plan-executer-pruefung
description: Die Standardschleife für nicht triviale Umsetzungsaufgaben — klären, planen, vom Executer-Subagent umsetzen lassen, selbst gegenprüfen, Befunde im Plan korrigieren, nachziehen lassen. Nutze diesen Skill, wenn ein Feature gebaut, ein Refactoring umgesetzt, eine mehrteilige Änderung ausgeführt oder ein bestehender Plan abgearbeitet werden soll, und immer dann, wenn von Executer, Subagent, Delegieren, Plan abarbeiten oder Gegenprüfen die Rede ist. Nicht für Einzeiler und nicht für offene Recherche.
---
# Standardschleife: Planen — Ausführen — Gegenprüfen
Rollen sind fest verteilt: **du (Opus) denkst und prüfst, der Executer tippt.** Vertausche das
nicht. Du schreibst keinen Produktivcode, solange die Schleife läuft; er trifft keine
Entscheidungen.
```
1 Klären → 2 Planen → 3 Ausführen → 4 Gegenprüfen → 5 Korrigieren → 6 Abschließen
du du Executer du du du
↑ │
└───── Befund ────┘
```
## Bevor du startest
**Lies `AI/Regelwerk/codeworkflow.md`, Abschnitt „Standardschleife", vollständig.** Dort steht
die verbindliche Fassung jedes Schritts: Planaufteilung ab drei Tasks, Modellwahl je Taskdatei,
das Pflichtprogramm der Gegenprüfung, fortsetzen oder frisch starten bei der Nachbesserung.
Diese Datei wiederholt das bewusst nicht. Eine zweite Fassung derselben Regel läuft aus dem
Ruder, sobald eine von beiden gepflegt wird — und die Kopie gewinnt, weil sie zuerst geladen
wird.
Für die Plankritik in Schritt 2, die Welle in Schritt 3 und die Prüfung in Schritt 4 gibt es
gespeicherte Workflows mit eingebauten Rollenparametern: `plankritik`, `executer-welle` und
`pruefung` unter `.claude/workflows/`, dazu `welle-mit-pruefung` für Schritt 3 und 4 in einem
Aufruf und `planung` für die Erstplanung vieler Taskdateien. Aufruf und Argumente stehen jeweils
in `whenToUse` im Kopf der Datei. Wie viele Opus-Agenten sie gleichzeitig starten, regelt
`codeworkflow.md`, „Verbrauch und Opus-Plätze". Sie ersetzen deine eigene Gegenprüfung nicht.
Danach je nach Aufgabe zusätzlich:
- `AI/Regelwerk/codedoku.md`, sobald Produktivdateien entstehen oder angefasst werden.
- `AI/Regelwerk/pruefwerkzeuge.md`, sobald ein Skript entsteht, das etwas **beurteilt** statt
etwas zu tun — und zwar vor der ersten Zeile Code.
**Nur für Claude Fable 5.1:** Läuft diese Sitzung auf Fable 5.1, lies dazu die Abschnitte 3, 8 und
15 in `AI/Regelwerk/fable-5-1.md`. Andere Modelle entscheiden selbst.
## Die zwei Dinge, die du nie delegierst
1. **Die Gegenprüfung in Schritt 4.** Ein Agentenbericht ist eine Behauptung, kein Beleg. Du
wiederholst den Testlauf selbst, liest die zentralen Artefakte selbst und fährst einen
eigenen Lauf mit echten Daten und eigenem Prüfcode. Der `pruefer`-Subagent (`opus`)
verdoppelt das, er ersetzt es nicht. Du fährst sie, während er läuft, nur lesend
(`codeworkflow.md`, Schritt 4).
2. **Den Commit.** Subagenten committen nie — paralleles `git commit` kollidiert am Index-Lock.
## Grenzen
- Einzeiler, Tippfehler, offensichtliche Korrekturen: selbst machen. Der Auftragstext kostet
mehr als die Änderung.
- Offene Recherche, Architekturentscheidungen, Debugging ohne Hypothese: kein Executer. Erst
klären und planen.
Melde nichts als fertig, was du nicht selbst hast laufen sehen. Was offen bleibt, wird offen
benannt — nicht weggelassen.
+55
View File
@@ -0,0 +1,55 @@
---
name: plane-abgleich
description: Gleicht die Checkbox-Aufgaben aus todo.md der Vault-Code-Projekte mit den Work Items der Plane-Instanz ab — in beide Richtungen, mit Konflikterkennung. Nutze diesen Skill, wenn von Plane, Work Items, Issues, Aufgabenabgleich, „todo nach Plane" oder „was ist in Plane offen" die Rede ist.
---
# Plane-Abgleich
Gleicht `todo.md` und Plane in beide Richtungen ab.
## Voraussetzungen prüfen
Setze `PLANE_URL`, `PLANE_API_KEY` und `PLANE_WORKSPACE` voraus. Fehlen sie, meldet das
Skript das verständlich — gib die Meldung weiter und brich ab. Frage niemals nach dem
Schlüssel und schreibe ihn nirgends hin.
Arbeitsverzeichnis: `D:\Claude\Code Projekte\active - Plane Abgleich\skill\scripts`
## Ablauf
1. **Immer zuerst der Probelauf.** Ohne `--anwenden` schreibt nichts:
```
python -m plane.cli
```
2. **Bericht lesen.** Je Aufgabe ANLEGEN, NACH PLANE, NACH VAULT, ANHAENGEN, ZEILE WEG,
ITEM WEG oder KONFLIKT.
3. **Fehlende Zuordnung klären.** Meldet der Bericht für ein Projekt, dass `plane_project`
fehlt:
- Plane-Projekte zeigen: `python -m plane.cli --projekte`
- Dem Anwender die Liste vorlegen und fragen, welches Projekt gemeint ist. Nicht raten.
- Gibt es kein passendes, anlegen lassen — erst nach ausdrücklicher Zustimmung:
`python -m plane.cli --projekt-anlegen "Name" --kennung KRZ --anwenden`
- Die UUID ins Frontmatter der `todo.md` schreiben.
4. **Konflikte vorlegen.** Bei jedem KONFLIKT beide Fassungen zeigen und fragen, welche gilt.
Nicht selbst entscheiden.
5. **Freigabe einholen und anwenden.** Erst nach ausdrücklichem „ja":
```
python -m plane.cli --anwenden --konflikt PLA-42=vault
```
6. **Ergebnis melden.** Was angelegt, was in welche Richtung übertragen, was offen blieb.
## Grenzen
- `--anwenden` niemals ungefragt ausführen. Es verändert Plane für alle Beteiligten.
- Konflikte ohne Entscheidung bleiben liegen — das ist beabsichtigt, nicht umgehen.
- Es wird nie gelöscht: keine Zeile, kein Work Item, kein Projekt. Entfernte Zeilen und
entfernte Work Items werden nur gemeldet.
- Projekte werden nur auf ausdrückliche Anforderung angelegt, nie nebenbei im Abgleich.
@@ -0,0 +1 @@
"""Plane-Abgleich."""
@@ -0,0 +1,142 @@
"""Drei-Wege-Vergleich zwischen todo.md, Plane und dem Stand des letzten Laufs."""
from __future__ import annotations
from dataclasses import dataclass
ANLEGEN = "ANLEGEN"
NACH_PLANE = "NACH_PLANE"
NACH_VAULT = "NACH_VAULT"
KONFLIKT = "KONFLIKT"
ANHAENGEN = "ANHAENGEN"
ZEILE_ENTFERNT = "ZEILE_ENTFERNT"
ITEM_ENTFERNT = "ITEM_ENTFERNT"
UNVERAENDERT = "UNVERAENDERT"
@dataclass(frozen=True)
class Aufgabe:
"""Eine Checkbox-Zeile aus todo.md."""
zeilennummer: int
text: str
erledigt: bool
marker: str | None
einrueckung: str = ""
@dataclass(frozen=True)
class WorkItem:
id: str
marker: str
titel: str
erledigt: bool
@dataclass(frozen=True)
class Basiseintrag:
marker: str
work_item_id: str
titel: str
erledigt: bool
@dataclass(frozen=True)
class Aktion:
art: str
marker: str | None
vault: Aufgabe | None = None
plane: WorkItem | None = None
basis: Basiseintrag | None = None
ERLEDIGT_GRUPPEN = ("completed", "cancelled")
ZIEL_GRUPPE_ERLEDIGT = "completed"
ZIEL_GRUPPE_OFFEN = "unstarted"
def status_fuer(states: list[dict], erledigt: bool) -> str:
"""Erste State-UUID der Zielgruppe, sortiert nach sequence."""
gruppe = ZIEL_GRUPPE_ERLEDIGT if erledigt else ZIEL_GRUPPE_OFFEN
passend = [s for s in states if s.get("group") == gruppe]
if not passend:
raise ValueError(
f"Das Projekt hat keinen Status der Gruppe „{gruppe}“. "
f"Vorhanden sind: {sorted({s.get('group') for s in states})}."
)
return min(passend, key=lambda s: s.get("sequence", 0))["id"]
def item_erledigt(state_gruppe: str) -> bool:
return state_gruppe in ERLEDIGT_GRUPPEN
def _abweichung(titel_a: str, erledigt_a: bool, titel_b: str, erledigt_b: bool) -> bool:
return titel_a != titel_b or erledigt_a != erledigt_b
def berechne(
aufgaben: list[Aufgabe],
work_items: list[WorkItem],
basis: dict[str, Basiseintrag],
) -> list[Aktion]:
"""Drei-Wege-Vergleich. Ruft nichts auf und schreibt nichts."""
aktionen: list[Aktion] = []
items_nach_marker = {item.marker: item for item in work_items}
gesehene_marker: set[str] = set()
for aufgabe in aufgaben:
if aufgabe.marker is None:
aktionen.append(Aktion(art=ANLEGEN, marker=None, vault=aufgabe))
continue
gesehene_marker.add(aufgabe.marker)
item = items_nach_marker.get(aufgabe.marker)
eintrag = basis.get(aufgabe.marker)
if item is None:
aktionen.append(
Aktion(art=ITEM_ENTFERNT, marker=aufgabe.marker, vault=aufgabe, basis=eintrag)
)
continue
if eintrag is None:
# Ohne Basis lässt sich nicht entscheiden, wer sich bewegt hat.
art = (
UNVERAENDERT
if not _abweichung(aufgabe.text, aufgabe.erledigt, item.titel, item.erledigt)
else KONFLIKT
)
aktionen.append(
Aktion(art=art, marker=aufgabe.marker, vault=aufgabe, plane=item)
)
continue
vault_geaendert = _abweichung(
aufgabe.text, aufgabe.erledigt, eintrag.titel, eintrag.erledigt
)
plane_geaendert = _abweichung(
item.titel, item.erledigt, eintrag.titel, eintrag.erledigt
)
if vault_geaendert and plane_geaendert:
art = KONFLIKT
elif vault_geaendert:
art = NACH_PLANE
elif plane_geaendert:
art = NACH_VAULT
else:
art = UNVERAENDERT
aktionen.append(
Aktion(art=art, marker=aufgabe.marker, vault=aufgabe, plane=item, basis=eintrag)
)
for item in work_items:
if item.marker in gesehene_marker:
continue
eintrag = basis.get(item.marker)
art = ZEILE_ENTFERNT if eintrag else ANHAENGEN
aktionen.append(Aktion(art=art, marker=item.marker, plane=item, basis=eintrag))
return aktionen
@@ -0,0 +1,143 @@
"""HTTP-Zugriff auf die Plane-API."""
from __future__ import annotations
import time
from typing import Any
import requests
from .konfiguration import Konfiguration
HOECHSTE_WARTEZEIT = 65.0
VERSUCHE_BEI_DROSSELUNG = 3
class ApiFehler(Exception):
"""Plane antwortet nicht wie erwartet."""
class PlaneApi:
"""Kapselt alle Zugriffe auf die Plane-API.
Es gibt bewusst je erlaubtem HTTP-Verb eine eigene Methode. Ein Verb als
Parameter durchzureichen würde löschende Aufrufe formulierbar machen.
"""
def __init__(self, konfiguration: Konfiguration, sitzung=None, schlaeft=time.sleep):
self._k = konfiguration
self._s = sitzung if sitzung is not None else requests.Session()
self._schlaeft = schlaeft
@property
def basis(self) -> str:
return f"{self._k.basis_url}/api/v1/workspaces/{self._k.workspace}"
def _kopf(self) -> dict[str, str]:
return {"X-API-Key": self._k.api_key, "Content-Type": "application/json"}
def _sende(self, verb_funktion, url: str, **kw):
antwort = None
for _ in range(VERSUCHE_BEI_DROSSELUNG):
antwort = verb_funktion(url, headers=self._kopf(), timeout=30, **kw)
if antwort.status_code != 429:
break
self._schlaeft(self._wartezeit(antwort))
return self._geprueft(antwort)
def _geprueft(self, antwort):
if antwort.status_code == 401:
raise ApiFehler(
"Plane lehnt den Schlüssel ab (401). PLANE_API_KEY prüfen."
)
if antwort.status_code == 403:
raise ApiFehler(
f"Plane verweigert den Zugriff (403). Entweder ist der Workspace-Slug "
f"„{self._k.workspace}“ falsch oder der Schlüssel hat dort "
f"keine Rechte — beides antwortet gleich."
)
if antwort.status_code == 429:
raise ApiFehler(
"Plane drosselt weiterhin (429), auch nach mehreren Versuchen. "
"Später erneut versuchen."
)
if antwort.status_code >= 400:
raise ApiFehler(
f"Plane antwortet {antwort.status_code}: {antwort.text[:300]}"
)
return antwort
@staticmethod
def _wartezeit(antwort, jetzt: float | None = None) -> float:
"""Liest X-RateLimit-Reset; der Wert kann Zeitstempel oder Restdauer sein."""
jetzt = time.time() if jetzt is None else jetzt
roh = antwort.headers.get("X-RateLimit-Reset", "")
try:
wert = float(roh)
except (TypeError, ValueError):
return 5.0
rest = wert - jetzt if wert > 1_000_000 else wert
return min(max(rest, 1.0), HOECHSTE_WARTEZEIT)
def _get(self, url: str, **kw):
return self._sende(self._s.get, url, **kw)
def _post(self, url: str, **kw):
return self._sende(self._s.post, url, **kw)
def _patch(self, url: str, **kw):
return self._sende(self._s.patch, url, **kw)
def _alle(self, url: str, params: dict[str, Any] | None = None) -> list[dict]:
"""Folgt der Cursor-Pagination bis zum Ende."""
params = dict(params or {})
params.setdefault("per_page", 100)
gesammelt: list[dict] = []
while True:
daten = self._get(url, params=params).json()
if isinstance(daten, list):
return daten
gesammelt.extend(daten.get("results", []))
if not daten.get("next_page_results"):
return gesammelt
cursor = daten.get("next_cursor")
if not cursor:
return gesammelt
params["cursor"] = cursor
# --- Fachmethoden -----------------------------------------------------
def projekte(self) -> list[dict]:
return self._alle(f"{self.basis}/projects/")
def projekt_anlegen(self, name: str, kennung: str) -> dict:
gesaeubert = "".join(z for z in kennung.upper() if z.isascii() and z.isalnum())
if not gesaeubert:
raise ApiFehler(
f"Die Kennung „{kennung}“ enthält kein einziges "
f"verwendbares Zeichen (erlaubt sind A-Z und 0-9)."
)
return self._post(
f"{self.basis}/projects/",
json={"name": name, "identifier": gesaeubert},
).json()
def states(self, projekt_id: str) -> list[dict]:
return self._alle(f"{self.basis}/projects/{projekt_id}/states/")
def work_items(self, projekt_id: str) -> list[dict]:
return self._alle(f"{self.basis}/projects/{projekt_id}/work-items/")
def work_item_anlegen(self, projekt_id: str, titel: str, state_id: str) -> dict:
return self._post(
f"{self.basis}/projects/{projekt_id}/work-items/",
json={"name": titel, "state": state_id},
).json()
def work_item_aendern(
self, projekt_id: str, work_item_id: str, felder: dict[str, Any]
) -> dict:
return self._patch(
f"{self.basis}/projects/{projekt_id}/work-items/{work_item_id}/",
json=felder,
).json()
@@ -0,0 +1,79 @@
"""Berichtsausgabe. Läuft auf der Windows-Konsole unter cp1252, nicht unter UTF-8."""
from __future__ import annotations
from .abgleich import (
ANHAENGEN,
ANLEGEN,
ITEM_ENTFERNT,
KONFLIKT,
NACH_PLANE,
NACH_VAULT,
UNVERAENDERT,
ZEILE_ENTFERNT,
Aktion,
)
BESCHRIFTUNG = {
ANLEGEN: "ANLEGEN ",
NACH_PLANE: "NACH PLANE ",
NACH_VAULT: "NACH VAULT ",
KONFLIKT: "KONFLIKT ",
ANHAENGEN: "ANHAENGEN ",
ZEILE_ENTFERNT: "ZEILE WEG ",
ITEM_ENTFERNT: "ITEM WEG ",
}
def sichere_ausgabe(strom, text: str) -> None:
"""Schreibt so, dass eine cp1252-Konsole den Lauf nicht abbricht."""
kodierung = getattr(strom, "encoding", None) or "utf-8"
try:
text.encode(kodierung)
except (UnicodeEncodeError, LookupError):
text = text.encode(kodierung, errors="replace").decode(kodierung, errors="replace")
strom.write(text + "\n")
def schreibe_bericht(strom, projektname: str, aktionen: list[Aktion], anwenden: bool) -> None:
sichere_ausgabe(strom, "")
sichere_ausgabe(strom, f"=== {projektname} ===")
kopf = "ANGEWENDET" if anwenden else "Probelauf - es wurde nichts geschrieben"
sichere_ausgabe(strom, kopf)
zaehler: dict[str, int] = {}
for aktion in aktionen:
zaehler[aktion.art] = zaehler.get(aktion.art, 0) + 1
if aktion.art == UNVERAENDERT:
continue
sichere_ausgabe(
strom,
f" {BESCHRIFTUNG.get(aktion.art, aktion.art)} "
f"{aktion.marker or '(neu)'} {_titel(aktion)}",
)
if aktion.art == KONFLIKT:
sichere_ausgabe(strom, f" Vault: [{_haken(aktion.vault.erledigt)}] "
f"{aktion.vault.text}")
sichere_ausgabe(strom, f" Plane: [{_haken(aktion.plane.erledigt)}] "
f"{aktion.plane.titel}")
sichere_ausgabe(
strom,
f" Entscheiden mit --konflikt {aktion.marker}=vault "
f"oder --konflikt {aktion.marker}=plane",
)
sichere_ausgabe(strom, " ---")
for art in sorted(zaehler):
sichere_ausgabe(strom, f" {BESCHRIFTUNG.get(art, art).strip()}: {zaehler[art]}")
def _titel(aktion: Aktion) -> str:
if aktion.vault is not None:
return aktion.vault.text
if aktion.plane is not None:
return aktion.plane.titel
return ""
def _haken(erledigt: bool) -> str:
return "x" if erledigt else " "
@@ -0,0 +1,309 @@
"""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))
@@ -0,0 +1,51 @@
"""Liest die Plane-Zugangsdaten aus Umgebungsvariablen."""
from __future__ import annotations
import os
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
STANDARD_QUELLE = Path(r"D:\Claude\Code Projekte")
PFLICHTVARIABLEN = ("PLANE_URL", "PLANE_API_KEY", "PLANE_WORKSPACE")
class KonfigurationsFehler(Exception):
"""Zugangsdaten fehlen oder sind unbrauchbar."""
@dataclass(frozen=True)
class Konfiguration:
basis_url: str
api_key: str = field(repr=False)
workspace: str = ""
quelle: Path = STANDARD_QUELLE
def lade_konfiguration(umgebung: Mapping[str, str] | None = None) -> Konfiguration:
"""Baut die Konfiguration aus der Umgebung; meldet alle Lücken auf einmal."""
werte = os.environ if umgebung is None else umgebung
fehlend = [name for name in PFLICHTVARIABLEN if not werte.get(name, "").strip()]
if fehlend:
raise KonfigurationsFehler(
"Diese Umgebungsvariablen fehlen oder sind leer: "
+ ", ".join(fehlend)
+ ". Siehe README.md, Abschnitt \u201eEinrichtung\u201c."
)
basis_url = werte["PLANE_URL"].strip().rstrip("/")
if not basis_url.startswith(("http://", "https://")):
raise KonfigurationsFehler(
f"PLANE_URL muss mit http:// oder https:// beginnen, "
f"ist aber \u201e{basis_url}\u201c."
)
quelle = werte.get("PLANE_QUELLE", "").strip()
return Konfiguration(
basis_url=basis_url,
api_key=werte["PLANE_API_KEY"].strip(),
workspace=werte["PLANE_WORKSPACE"].strip(),
quelle=Path(quelle) if quelle else STANDARD_QUELLE,
)
@@ -0,0 +1,130 @@
"""Liest und schreibt die todo.md eines Vault-Projekts."""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from .abgleich import Aufgabe
CHECKBOX_MUSTER = re.compile(r"^(?P<einrueckung>[ \t]*)- \[(?P<haken>[ xX])\] (?P<rest>.*)$")
MARKER_MUSTER = re.compile(r"\s*<!--\s*plane:(?P<marker>[A-Z0-9]+-\d+)\s*-->\s*$")
ABSCHNITT_AUS_PLANE = "## Aus Plane"
@dataclass
class TodoDatei:
pfad: Path
zeilen: list[str]
frontmatter: dict[str, str]
aufgaben: list[Aufgabe]
frontmatter_zeilen: int
def _lies_frontmatter(zeilen: list[str]) -> tuple[dict[str, str], int]:
"""Gibt die flachen Schlüssel-Wert-Paare und die Anzahl belegter Zeilen zurück."""
if not zeilen or zeilen[0].strip() != "---":
return {}, 0
for index in range(1, len(zeilen)):
if zeilen[index].strip() == "---":
werte: dict[str, str] = {}
for zeile in zeilen[1:index]:
schluessel, trenner, wert = zeile.partition(":")
if trenner:
werte[schluessel.strip()] = wert.strip()
return werte, index + 1
return {}, 0
def lies(pfad: Path) -> TodoDatei:
# utf-8-sig statt utf-8: ein BOM am Dateianfang wuerde sonst als  in
# Zeile 0 stehen bleiben, das Frontmatter waere unsichtbar und der Skill
# meldete "plane_project fehlt", obwohl es dasteht. Windows-Editoren und
# PowerShell 5.1 schreiben UTF-8 regelmaessig mit BOM. Geschrieben wird
# weiterhin ohne BOM.
text = Path(pfad).read_text(encoding="utf-8-sig")
zeilen = text.split("\n")
frontmatter, belegt = _lies_frontmatter(zeilen)
aufgaben: list[Aufgabe] = []
for nummer, zeile in enumerate(zeilen):
if nummer < belegt:
continue
treffer = CHECKBOX_MUSTER.match(zeile)
if not treffer:
continue
rest = treffer.group("rest")
marker_treffer = MARKER_MUSTER.search(rest)
marker = marker_treffer.group("marker") if marker_treffer else None
text_ohne_marker = MARKER_MUSTER.sub("", rest).strip() if marker else rest.strip()
aufgaben.append(
Aufgabe(
zeilennummer=nummer,
text=text_ohne_marker,
erledigt=treffer.group("haken").lower() == "x",
marker=marker,
einrueckung=treffer.group("einrueckung"),
)
)
return TodoDatei(
pfad=Path(pfad),
zeilen=zeilen,
frontmatter=frontmatter,
aufgaben=aufgaben,
frontmatter_zeilen=belegt,
)
def plane_projekt(frontmatter: dict[str, str]) -> str | None:
wert = frontmatter.get("plane_project", "").strip()
return wert or None
def baue_zeile(einrueckung: str, text: str, erledigt: bool, marker: str | None) -> str:
haken = "x" if erledigt else " "
zeile = f"{einrueckung}- [{haken}] {text}"
if marker:
zeile = f"{zeile} <!-- plane:{marker} -->"
return zeile
def schreibe(
datei: TodoDatei,
aenderungen: dict[int, tuple[str, bool, str | None]],
neue_zeilen: list[str],
) -> str:
"""Baut den neuen Dateiinhalt. Unberührte Zeilen bleiben byteweise gleich."""
zeilen = list(datei.zeilen)
nach_nummer = {aufgabe.zeilennummer: aufgabe for aufgabe in datei.aufgaben}
for nummer, (text, erledigt, marker) in aenderungen.items():
alt = nach_nummer.get(nummer)
einrueckung = alt.einrueckung if alt else ""
zeilen[nummer] = baue_zeile(einrueckung, text, erledigt, marker)
if neue_zeilen:
zeilen = _haenge_an_aus_plane(zeilen, neue_zeilen)
return "\n".join(zeilen)
def _haenge_an_aus_plane(zeilen: list[str], neue_zeilen: list[str]) -> list[str]:
ergebnis = list(zeilen)
if ABSCHNITT_AUS_PLANE in [z.strip() for z in ergebnis]:
letzte = max(
i for i, z in enumerate(ergebnis) if z.strip() == ABSCHNITT_AUS_PLANE
)
einfuegen_ab = len(ergebnis)
for i in range(letzte + 1, len(ergebnis)):
if ergebnis[i].startswith("#"):
einfuegen_ab = i
break
while einfuegen_ab > letzte + 1 and not ergebnis[einfuegen_ab - 1].strip():
einfuegen_ab -= 1
return ergebnis[:einfuegen_ab] + neue_zeilen + ergebnis[einfuegen_ab:]
while ergebnis and not ergebnis[-1].strip():
ergebnis.pop()
return ergebnis + ["", ABSCHNITT_AUS_PLANE, ""] + neue_zeilen + [""]
@@ -0,0 +1,77 @@
"""Der Stand des letzten Laufs — die Basis des Drei-Wege-Vergleichs."""
from __future__ import annotations
import json
from datetime import datetime
from pathlib import Path
from .abgleich import Basiseintrag
DATEINAME = ".plane-sync.json"
VERSION = 1
class Zustand:
def __init__(self, pfad: Path, roh: dict, warnung: str | None = None):
self.pfad = pfad
self.warnung = warnung
self.plane_projekt: str | None = roh.get("plane_project")
self.eintraege: dict[str, Basiseintrag] = {
marker: Basiseintrag(
marker=marker,
work_item_id=werte["work_item_id"],
titel=werte["titel"],
erledigt=bool(werte["erledigt"]),
)
for marker, werte in (roh.get("eintraege") or {}).items()
}
self._letzter_lauf: str | None = roh.get("letzter_lauf")
def setze(self, eintrag: Basiseintrag) -> None:
"""Trägt einen Eintrag ein und speichert sofort — das macht den Lauf wiederaufsetzbar."""
self.eintraege[eintrag.marker] = eintrag
self.speichere()
def entferne_marker(self, marker: str) -> None:
self.eintraege.pop(marker, None)
self.speichere()
def setze_lauf_zeit(self) -> None:
self._letzter_lauf = datetime.now().astimezone().isoformat(timespec="seconds")
self.speichere()
def speichere(self) -> None:
inhalt = {
"version": VERSION,
"plane_project": self.plane_projekt,
"letzter_lauf": self._letzter_lauf,
"eintraege": {
marker: {
"work_item_id": e.work_item_id,
"titel": e.titel,
"erledigt": e.erledigt,
}
for marker, e in sorted(self.eintraege.items())
},
}
self.pfad.write_text(
json.dumps(inhalt, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
def lies(pfad: Path) -> Zustand:
pfad = Path(pfad)
if not pfad.exists():
return Zustand(pfad, {})
try:
return Zustand(pfad, json.loads(pfad.read_text(encoding="utf-8")))
except (json.JSONDecodeError, KeyError, TypeError) as fehler:
return Zustand(
pfad,
{},
warnung=(
f"{pfad} ist unlesbar ({fehler}). Der Lauf behandelt sie als leer; "
"dadurch erscheinen bereits abgeglichene Aufgaben erneut."
),
)