Regelwerk, Agenten, Skills und Werkzeuge veröffentlichen
Erster Stand des öffentlichen Ausschnitts aus dem Vault. Die .gitignore arbeitet als Positivliste: ausgeschlossen ist alles, eingeschlossen nur, was ausdrücklich wieder aufgenommen wird. Notizen, Projekte, Finanzen, das Sitzungsgedächtnis und settings.local.json bleiben damit außen vor. Der Kundenname in AI/Regelwerk/seo.md ist durch eine neutrale Bezeichnung ersetzt; die Regeln selbst sind unverändert. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
commit
c064c729ed
82 files changed
+13428
No files matched your search
@@ -0,0 +1,56 @@
|
||||
---
|
||||
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, 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 Notiz ANLEGEN, AKTUALISIEREN, ÜBERSPRUNGEN oder FEHLER.
|
||||
|
||||
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.
|
||||
|
||||
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.
|
||||
|
||||
## 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 werden zu echten Seitenlinks. Verweise auf
|
||||
noch nicht hochgeladene Notizen bleiben Klartext und werden im Bericht gemeldet.
|
||||
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,41 @@
|
||||
"""Findet im Quellordner die Notizen, die zum Upload freigegeben sind."""
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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 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 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,106 @@
|
||||
"""Findet Bilder im Vault und lädt sie in die BookStack-Bildergalerie."""
|
||||
|
||||
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 _vorhandene(self, seiten_id: int) -> dict[str, str]:
|
||||
"""Bereits an dieser Seite hängende Galeriebilder, Name → URL."""
|
||||
try:
|
||||
eintraege = self._api.liste(f"/api/image-gallery?filter[uploaded_to]={seiten_id}")
|
||||
except ApiFehler:
|
||||
return {}
|
||||
return {str(eintrag.get("name", "")): str(eintrag.get("url", "")) for eintrag in eintraege}
|
||||
|
||||
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
|
||||
|
||||
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,228 @@
|
||||
"""CLI: findet freigegebene Notizen und spielt sie nach BookStack."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from bookstack.abgleich import Aktion, berechne_pruefsumme, entscheide
|
||||
from bookstack.api import ApiFehler, BookStackApi
|
||||
from bookstack.auswahl import finde_kandidaten, ist_freigegeben, titel_von
|
||||
from bookstack.bilder import Bilduploader
|
||||
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 baue_index
|
||||
|
||||
|
||||
@dataclass
|
||||
class Ergebnis:
|
||||
pfad: Path
|
||||
aktion: str
|
||||
text: str
|
||||
|
||||
|
||||
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 verarbeite(
|
||||
notiz: Notiz,
|
||||
aufloeser: Strukturaufloeser,
|
||||
api: BookStackApi,
|
||||
*,
|
||||
anwenden: bool,
|
||||
verweise: dict[str, str],
|
||||
bilduploader: Bilduploader,
|
||||
) -> Ergebnis:
|
||||
titel = titel_von(notiz)
|
||||
rumpf = ohne_titelueberschrift(notiz.rumpf, titel)
|
||||
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(titel, umwandlung.markdown, ziel.buch_id, ziel.kapitel_id)
|
||||
entscheidung = entscheide(notiz, pruefsumme)
|
||||
|
||||
beschreibung = f"{titel} → Buch-ID {ziel.buch_id}"
|
||||
if ziel.kapitel_id:
|
||||
beschreibung += f", Kapitel-ID {ziel.kapitel_id}"
|
||||
# Die Bildhinweise aus Pass 1 sagen nur, dass zu diesem Zeitpunkt noch keine URL
|
||||
# vorlag — nicht, dass das Bild fehlschlägt. Was wirklich passiert ist, meldet
|
||||
# weiter unten der Bilduploader (bzw. im Probelauf „würde hochgeladen").
|
||||
for offen in umwandlung.offene_verweise:
|
||||
beschreibung += f"\n Verweis nicht auflösbar: {offen}"
|
||||
for angelegt in ziel.angelegt:
|
||||
beschreibung += f"\n Angelegt: {angelegt}"
|
||||
|
||||
if entscheidung.aktion is Aktion.UEBERSPRINGEN:
|
||||
return Ergebnis(notiz.pfad, "ÜBERSPRUNGEN", entscheidung.begruendung)
|
||||
if not anwenden:
|
||||
for name in umwandlung.bildnamen:
|
||||
beschreibung += f"\n Bild würde hochgeladen: {name}"
|
||||
return Ergebnis(notiz.pfad, entscheidung.aktion.name, beschreibung)
|
||||
|
||||
def nutzlast(markdown: str) -> dict[str, object]:
|
||||
daten: dict[str, object] = {"name": titel, "markdown": markdown}
|
||||
if ziel.kapitel_id:
|
||||
daten["chapter_id"] = ziel.kapitel_id
|
||||
else:
|
||||
daten["book_id"] = ziel.buch_id
|
||||
return daten
|
||||
|
||||
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))
|
||||
beschreibung += "\n Hinweis: Seite war gelöscht, neu angelegt."
|
||||
else:
|
||||
antwort = api.anfrage("POST", "/api/pages", nutzlast(umwandlung.markdown))
|
||||
|
||||
seiten_id = int(antwort["id"])
|
||||
|
||||
if umwandlung.bildnamen:
|
||||
tabelle, bildhinweise = bilduploader.lade_hoch(
|
||||
notiz.pfad, seiten_id, umwandlung.bildnamen
|
||||
)
|
||||
for hinweis in bildhinweise:
|
||||
beschreibung += f"\n Hinweis: {hinweis}"
|
||||
if tabelle:
|
||||
zweiter = konvertiere(rumpf, verweise=verweise, bilder=tabelle)
|
||||
api.anfrage("PUT", f"/api/pages/{seiten_id}", nutzlast(zweiter.markdown))
|
||||
beschreibung += f"\n Bilder übertragen: {len(tabelle)}"
|
||||
|
||||
schreibe_felder(
|
||||
notiz.pfad,
|
||||
{
|
||||
"bookstack_page_id": str(seiten_id),
|
||||
"bookstack_hash": pruefsumme,
|
||||
"bookstack_synced": datetime.now().astimezone().isoformat(timespec="seconds"),
|
||||
},
|
||||
)
|
||||
return Ergebnis(notiz.pfad, entscheidung.aktion.name, beschreibung)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
if einstellungen.baum:
|
||||
try:
|
||||
print(aufloeser.baum())
|
||||
except ApiFehler as fehler:
|
||||
print(f"Fehler: {fehler}")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
if einstellungen.nur:
|
||||
notiz = lade_notiz(einstellungen.nur)
|
||||
kandidaten = [notiz] if ist_freigegeben(notiz) else []
|
||||
if not kandidaten:
|
||||
print(f"{einstellungen.nur} trägt kein #review-done — nichts zu tun.")
|
||||
return 0
|
||||
else:
|
||||
kandidaten = finde_kandidaten(konfiguration.quelle)
|
||||
|
||||
kopf = "BookStack-Upload" + ("" if einstellungen.anwenden else " (Probelauf)")
|
||||
print(f"{kopf} — Quelle: {konfiguration.quelle}")
|
||||
if not kandidaten:
|
||||
print("Keine Notiz mit #review-done gefunden.")
|
||||
return 0
|
||||
|
||||
verweise = baue_index(konfiguration.quelle, konfiguration.basis_url)
|
||||
|
||||
ergebnisse: list[Ergebnis] = []
|
||||
fehler_anzahl = 0
|
||||
for notiz in kandidaten:
|
||||
try:
|
||||
ergebnisse.append(
|
||||
verarbeite(
|
||||
notiz,
|
||||
aufloeser,
|
||||
api,
|
||||
anwenden=einstellungen.anwenden,
|
||||
verweise=verweise,
|
||||
bilduploader=bilduploader,
|
||||
)
|
||||
)
|
||||
except (StrukturFehler, ApiFehler, OSError) as fehler:
|
||||
fehler_anzahl += 1
|
||||
ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler)))
|
||||
|
||||
for ergebnis in ergebnisse:
|
||||
try:
|
||||
anzeige = ergebnis.pfad.relative_to(konfiguration.quelle)
|
||||
except ValueError:
|
||||
anzeige = ergebnis.pfad
|
||||
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 einstellungen.anwenden:
|
||||
print("Probelauf — nichts geschrieben. Mit --anwenden ausführen.")
|
||||
return 1 if fehler_anzahl 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,103 @@
|
||||
"""Übersetzt Obsidian-Eigenheiten in Markdown, das BookStack versteht."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
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)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Sammler:
|
||||
verweise: Mapping[str, str]
|
||||
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)
|
||||
|
||||
|
||||
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 = treffer.group(1).strip()
|
||||
if name not in sammler.bildnamen:
|
||||
sammler.bildnamen.append(name)
|
||||
url = sammler.bilder.get(name)
|
||||
if url:
|
||||
return f""
|
||||
sammler.hinweise.append(f"Bild nicht übertragen: {name}")
|
||||
return f"*(Bild „{name}“ wurde nicht übertragen)*"
|
||||
|
||||
def link(treffer: re.Match[str]) -> str:
|
||||
ziel, alias = treffer.group(1), treffer.group(2)
|
||||
kurz = ziel.split("/")[-1].split("#")[0].strip()
|
||||
text_teil = (alias or kurz).strip()
|
||||
url = sammler.verweise.get(kurz.casefold())
|
||||
if url:
|
||||
return f"[{text_teil}]({url})"
|
||||
if kurz not in sammler.offene_verweise:
|
||||
sammler.offene_verweise.append(kurz)
|
||||
return text_teil
|
||||
|
||||
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,
|
||||
bilder: Mapping[str, str] | None = None,
|
||||
) -> Umwandlung:
|
||||
sammler = Sammler(verweise=verweise or {}, 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,
|
||||
)
|
||||
@@ -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,156 @@
|
||||
"""Löst Regal-, Buch- und Kapitelnamen in IDs auf und legt Fehlendes auf Wunsch an."""
|
||||
|
||||
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
|
||||
kapitel_id: int | None = None
|
||||
angelegt: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
def _passt(eintrag: dict, name: str) -> bool:
|
||||
return str(eintrag.get("name", "")).strip().casefold() == name.strip().casefold()
|
||||
|
||||
|
||||
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]] = {}
|
||||
|
||||
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 _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)
|
||||
angelegt.append(self._vermerke("Regal", regal_name))
|
||||
if self._nur_planen:
|
||||
return
|
||||
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)
|
||||
angelegt.append(self._vermerke("Buch", buch))
|
||||
if self._nur_planen:
|
||||
gefunden = {"id": 0}
|
||||
else:
|
||||
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)
|
||||
angelegt.append(self._vermerke("Kapitel", kapitel))
|
||||
if self._nur_planen:
|
||||
treffer = {"id": 0}
|
||||
else:
|
||||
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,34 @@
|
||||
"""Baut den Index bereits hochgeladener Notizen für die Wikilink-Auflösung."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
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 baue_index(quelle: Path, basis_url: str) -> dict[str, str]:
|
||||
"""Dateiname ohne Endung (kleingeschrieben) → Permalink."""
|
||||
if not quelle.is_dir():
|
||||
return {}
|
||||
index: dict[str, str] = {}
|
||||
for pfad in sorted(quelle.rglob("*.md")):
|
||||
if pfad.name.startswith((".", "_")):
|
||||
continue
|
||||
roh_id = lade_notiz(pfad).frontmatter.get("bookstack_page_id", "").strip()
|
||||
try:
|
||||
seiten_id = int(roh_id)
|
||||
except ValueError:
|
||||
continue
|
||||
index[pfad.stem.casefold()] = permalink(basis_url, seiten_id)
|
||||
return index
|
||||
@@ -0,0 +1,54 @@
|
||||
---
|
||||
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.
|
||||
|
||||
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.
|
||||
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.
|
||||
@@ -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."
|
||||
),
|
||||
)
|
||||
Reference in new issue
Block a user