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:
Pascal MichelandClaude Opus 5 committed 2026-09-17 22:38:15 +02:00
commit 2c9a4475fc
97 files changed
+14627

No files matched your search

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"![{name}]({url})"
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