Pascal MichelandClaude Opus 5
c064c729ed
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]>
229 lines
8.3 KiB
Python
229 lines
8.3 KiB
Python
"""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())
|