"""Kommandozeile: lädt freigegebene Notizen und Beschreibungsdateien nach BookStack, zieht Verweise nach und archiviert. Ersteller: Pascal Michel Erstellt: 2026-08-05 Doku: docs/code/hochladen.md · Version und Änderungen: CHANGELOG.md """ from __future__ import annotations import argparse import sys from datetime import datetime from pathlib import Path from bookstack.abgleich import Aktion, berechne_pruefsumme, entscheide from bookstack.api import ApiFehler, BookStackApi from bookstack.archiv import ARCHIV_ORDNERNAME, archiviere, ist_archiviert from bookstack.auswahl import ( beschreibungsart, finde_beschreibungen, finde_kandidaten, ist_freigegeben, titel_von, ) from bookstack.beschreibung import ( UEBERSICHTSTITEL, BeschreibungsFehler, beschreibungs_pruefsumme, pruefe_beschreibung_html, pruefe_ziel_felder, setze_beschreibung, vorspann, zu_beschreibungs_html, ) from bookstack.bilder import Bilduploader from bookstack.ergebnis import ( ARCHIVIERBAR, Beschreibungsstand, Ergebnis, Seitenstand, seiten_nutzlast, ) from bookstack.konfiguration import KonfigurationsFehler, lade_konfiguration from bookstack.konvertierung import konvertiere, ohne_titelueberschrift from bookstack.notiz import Notiz, lade_notiz, schreibe_felder from bookstack.struktur import StrukturFehler, Strukturaufloeser from bookstack.verweise import Verweisindex, baue_index from bookstack.verweisrunde import ( baue_endindex, fuehre_verweisrunde, haenge_offene_an, plane_endindex, ) # Fehler je Datei in Durchgang 1: Sie werden zur FEHLER-Zeile, der Lauf geht weiter. DATEIFEHLER = (StrukturFehler, ApiFehler, OSError, BeschreibungsFehler) def _stelle_ausgabe_um() -> None: """Windows-Konsolen laufen oft unter cp1252 — „→" bricht den Lauf sonst ab. Gefunden beim Lauf mit echten Daten: pytest fängt stdout in UTF-8 ab, deshalb bleibt der Fehler in der Testsuite unsichtbar und schlägt erst beim Anwender zu. """ for strom in (sys.stdout, sys.stderr): umschalten = getattr(strom, "reconfigure", None) if umschalten is None: continue try: umschalten(encoding="utf-8", errors="replace") except (ValueError, OSError): pass def _argumente(argumente: list[str]) -> argparse.Namespace: zerleger = argparse.ArgumentParser( prog="bookstack.hochladen", description="Lädt freigegebene Vault-Notizen (#review-done) nach BookStack.", ) zerleger.add_argument("--anwenden", action="store_true", help="tatsächlich schreiben") zerleger.add_argument( "--struktur-anlegen", action="store_true", help="fehlende Regale/Bücher/Kapitel anlegen" ) zerleger.add_argument("--nur", type=Path, help="nur diese eine Datei verarbeiten") zerleger.add_argument("--baum", action="store_true", help="nur die BookStack-Struktur zeigen") return zerleger.parse_args(argumente) def _jetzt() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def _anzeige(pfad: Path, quelle: Path) -> Path: """Pfad für den Bericht: relativ zur Quelle, sonst wie er ist.""" try: return pfad.relative_to(quelle) except ValueError: return pfad def _mit_art(notizen: list[Notiz]) -> list[tuple[Notiz, str]]: """Beschreibungsdateien samt Art ('buch' oder 'kapitel'); andere Notizen fallen heraus.""" paare: list[tuple[Notiz, str]] = [] for notiz in notizen: art = beschreibungsart(notiz.pfad) if art is not None: paare.append((notiz, art)) return paare def verarbeite( notiz: Notiz, aufloeser: Strukturaufloeser, api: BookStackApi, *, anwenden: bool, verweise: Verweisindex, bilduploader: Bilduploader, titel: str | None = None, prioritaet: int | None = None, ) -> Ergebnis: """Durchgang 1 für eine Seite: anlegen, aktualisieren oder überspringen. Das Ergebnis trägt bei jeder dieser Aktionen einen Seitenstand, auch im Probelauf. Offene Verweise hängt erst haenge_offene_an an, nach der Verweisrunde. """ seitentitel = titel_von(notiz) if titel is None else titel rumpf = ohne_titelueberschrift(notiz.rumpf, seitentitel) umwandlung = konvertiere(rumpf, verweise=verweise) ziel = aufloeser.aufloesen( notiz.frontmatter.get("bookstack_shelf"), notiz.frontmatter.get("bookstack_book", ""), notiz.frontmatter.get("bookstack_chapter"), ) # Bewusst ueber den Pass-1-Markdown: Bild-URLs haengen an der Seiten-ID, # nicht am Inhalt, und wuerden sonst bei jedem Lauf ein Update ausloesen. pruefsumme = berechne_pruefsumme( seitentitel, umwandlung.markdown, ziel.buch_id, ziel.kapitel_id ) entscheidung = entscheide(notiz, pruefsumme) def stand(seiten_id: int | None, bilder: dict[str, str] | None) -> Seitenstand: return Seitenstand( titel=seitentitel, rumpf=rumpf, ziel=ziel, seiten_id=seiten_id, prioritaet=prioritaet, markdown=umwandlung.markdown, aufgeloeste_verweise=umwandlung.aufgeloeste_verweise, offene_verweise=umwandlung.offene_verweise, bildnamen=umwandlung.bildnamen, bilder=bilder, ) text = f"{seitentitel} → Buch-ID {ziel.buch_id}" # is not None: Die Platzhalter-ID 0 eines geplanten Kapitels erscheint im Probelauf. if ziel.kapitel_id is not None: text += f", Kapitel-ID {ziel.kapitel_id}" for angelegt in ziel.angelegt: text += f"\n Angelegt: {angelegt}" if entscheidung.aktion is Aktion.UEBERSPRINGEN: return Ergebnis( notiz.pfad, "ÜBERSPRUNGEN", entscheidung.begruendung, seite=stand(entscheidung.seiten_id, None), ) if not anwenden: # Die Bildhinweise aus Pass 1 sagen nur, dass noch keine URL vorlag; im Probelauf # steht deshalb „würde hochgeladen“, was wirklich passiert, meldet der Bilduploader. for name in umwandlung.bildnamen: text += f"\n Bild würde hochgeladen: {name}" return Ergebnis( notiz.pfad, entscheidung.aktion.name, text, seite=stand(entscheidung.seiten_id, None), ) def nutzlast(markdown: str) -> dict[str, object]: return seiten_nutzlast(seitentitel, ziel, markdown, prioritaet) if entscheidung.aktion is Aktion.AKTUALISIEREN: try: antwort = api.anfrage( "PUT", f"/api/pages/{entscheidung.seiten_id}", nutzlast(umwandlung.markdown) ) except ApiFehler as fehler: if fehler.status != 404: raise # Seite wurde serverseitig gelöscht — neu anlegen statt scheitern. antwort = api.anfrage("POST", "/api/pages", nutzlast(umwandlung.markdown)) text += "\n Hinweis: Seite war gelöscht, neu angelegt." else: antwort = api.anfrage("POST", "/api/pages", nutzlast(umwandlung.markdown)) seiten_id = int(antwort["id"]) # Die ID steht sofort im Frontmatter: Scheitert danach der Bildteil, aktualisiert der nächste # Lauf diese Seite, statt sie ein zweites Mal anzulegen. Der Hash folgt erst am Ende, damit # der nächste Lauf eine halb fertige Seite nicht für fertig hält. schreibe_felder(notiz.pfad, {"bookstack_page_id": str(seiten_id)}) # Die Tabelle geht in den Seitenstand; die Verweisrunde sendet nie ohne sie. bilder: dict[str, str] = {} if umwandlung.bildnamen: bilder, bildhinweise = bilduploader.lade_hoch( notiz.pfad, seiten_id, umwandlung.bildnamen ) for hinweis in bildhinweise: text += f"\n Hinweis: {hinweis}" if bilder: zweiter = konvertiere(rumpf, verweise=verweise, bilder=bilder) api.anfrage("PUT", f"/api/pages/{seiten_id}", nutzlast(zweiter.markdown)) text += f"\n Bilder übertragen: {len(bilder)}" schreibe_felder( notiz.pfad, { "bookstack_page_id": str(seiten_id), "bookstack_hash": pruefsumme, "bookstack_synced": _jetzt(), }, ) return Ergebnis(notiz.pfad, entscheidung.aktion.name, text, seite=stand(seiten_id, bilder)) def verarbeite_beschreibung( notiz: Notiz, art: str, aufloeser: Strukturaufloeser, api: BookStackApi, *, anwenden: bool, verweise: Verweisindex, bilduploader: Bilduploader, ) -> list[Ergebnis]: """Durchgang 1 für eine Beschreibungsdatei: erst die Übersichtsseite, dann die Beschreibung. Fehler in pruefe_ziel_felder und im Seitenteil gehen an fuehre_aus. Fehler im Beschreibungsteil werden hier zur FEHLER-Zeile; die Seitenzeile bleibt stehen. """ pruefe_ziel_felder(art, notiz.frontmatter) seitenzeile = verarbeite( notiz, aufloeser, api, anwenden=anwenden, verweise=verweise, bilduploader=bilduploader, titel=UEBERSICHTSTITEL[art], prioritaet=0, ) ergebnisse = [seitenzeile] # verarbeite liefert bei ANLEGEN, AKTUALISIEREN und ÜBERSPRUNGEN immer einen Seitenstand. assert seitenzeile.seite is not None ziel = seitenzeile.seite.ziel ziel_id = ziel.buch_id if art == "buch" else ziel.kapitel_id kennung = f"Buch-ID {ziel.buch_id}" if art != "buch": kennung += f", Kapitel-ID {ziel.kapitel_id}" try: if ziel_id is None: raise BeschreibungsFehler("bookstack_chapter fehlt in der Kapitelbeschreibung") text = vorspann(notiz.rumpf) umgewandelt = zu_beschreibungs_html(text, verweise) pruefe_beschreibung_html(umgewandelt.html) pruefsumme = beschreibungs_pruefsumme(ziel_id, umgewandelt.html) stand = Beschreibungsstand( art=art, ziel_id=ziel_id, vorspann=text, html=umgewandelt.html, aufgeloeste_verweise=umgewandelt.aufgeloeste_verweise, ) # unverändert gilt auch im Probelauf; gesendet wird nur mit --anwenden und bei Abweichung. if notiz.frontmatter.get("bookstack_beschreibung_hash", "").strip() == pruefsumme: meldung = f"unverändert → {kennung}" elif not anwenden: meldung = f"würde gesetzt → {kennung}" else: setze_beschreibung(api, art, ziel_id, umgewandelt.html) schreibe_felder(notiz.pfad, {"bookstack_beschreibung_hash": pruefsumme}) meldung = f"gesetzt → {kennung}" except (BeschreibungsFehler, ApiFehler, OSError) as fehler: ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler))) return ergebnisse ergebnisse.append(Ergebnis(notiz.pfad, "BESCHREIBUNG", meldung, beschreibung=stand)) return ergebnisse def _archiviere(gruppen: dict[Path, list[Ergebnis]], quelle: Path, *, anwenden: bool) -> int: """Verschiebt je Pfad einmal ins Archiv; Rückgabe: Zahl der ARCHIV-FEHLER. Archiviert wird ein Pfad mit einer Zeile aus ARCHIVIERBAR und ohne FEHLER-Zeile, also nur, wenn Seite, Beschreibung und Verweisrunde gelungen sind. """ archiviert = 0 fehler_anzahl = 0 for pfad, zeilen in gruppen.items(): aktionen = {ergebnis.aktion for ergebnis in zeilen} if "FEHLER" in aktionen or aktionen.isdisjoint(ARCHIVIERBAR): continue try: relativ = pfad.relative_to(quelle) except ValueError: continue # z. B. --nur mit einem Pfad außerhalb der Quelle — nicht archivieren. if ist_archiviert(pfad, quelle): continue if not anwenden: print(f" würde archiviert: {relativ}") continue try: neu = archiviere(pfad, quelle) except OSError as fehler: # Archivkollision oder gesperrte Datei: Die Notiz bleibt liegen, der Lauf geht weiter. fehler_anzahl += 1 print(f" ARCHIV-FEHLER {relativ}\n {fehler}") continue archiviert += 1 print(f" ARCHIVIERT {relativ} → {neu.relative_to(quelle)}") if archiviert: print(f"{archiviert}× ins Archiv verschoben ({ARCHIV_ORDNERNAME})") return fehler_anzahl def fuehre_aus(argumente: list[str]) -> int: _stelle_ausgabe_um() einstellungen = _argumente(argumente) try: konfiguration = lade_konfiguration() except KonfigurationsFehler as fehler: print(f"Konfigurationsfehler: {fehler}") return 1 api = BookStackApi( konfiguration.basis_url, konfiguration.token_id, konfiguration.token_secret ) aufloeser = Strukturaufloeser( api, anlegen_erlaubt=einstellungen.struktur_anlegen, nur_planen=not einstellungen.anwenden, ) bilduploader = Bilduploader(api, konfiguration.vault) quelle = konfiguration.quelle anwenden = einstellungen.anwenden if einstellungen.baum: try: print(aufloeser.baum()) except ApiFehler as fehler: print(f"Fehler: {fehler}") return 1 return 0 # Auswahl: --nur ordnet die eine Datei über ihren Namen als Beschreibung oder Seite ein. if einstellungen.nur: notiz = lade_notiz(einstellungen.nur) if not ist_freigegeben(notiz): print(f"{einstellungen.nur} trägt kein #review-done — nichts zu tun.") return 0 beschreibungen = _mit_art([notiz]) kandidaten = [] if beschreibungen else [notiz] else: beschreibungen = _mit_art(finde_beschreibungen(quelle)) kandidaten = finde_kandidaten(quelle) kopf = "BookStack-Upload" + ("" if anwenden else " (Probelauf)") print(f"{kopf} — Quelle: {quelle}") if not beschreibungen and not kandidaten: print("Keine Notiz mit #review-done gefunden.") return 0 verweise = baue_index(quelle, konfiguration.basis_url, konfiguration.vault) # Durchgang 1: erst Beschreibungsdateien (Bücher vor Kapiteln), dann Seiten; der Index bleibt fest. ergebnisse: list[Ergebnis] = [] for notiz, art in beschreibungen: try: ergebnisse.extend( verarbeite_beschreibung( notiz, art, aufloeser, api, anwenden=anwenden, verweise=verweise, bilduploader=bilduploader, ) ) except DATEIFEHLER as fehler: ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler))) for notiz in kandidaten: try: ergebnisse.append( verarbeite( notiz, aufloeser, api, anwenden=anwenden, verweise=verweise, bilduploader=bilduploader, ) ) except DATEIFEHLER as fehler: ergebnisse.append(Ergebnis(notiz.pfad, "FEHLER", str(fehler))) # Verweisrunde nur mit --anwenden; der Probelauf plant den Endindex nur und sendet nichts. if anwenden: endindex = baue_endindex( verweise, ergebnisse, quelle=quelle, vault=konfiguration.vault, basis_url=konfiguration.basis_url, ) ergebnisse += fuehre_verweisrunde(ergebnisse, endindex, api, bilduploader) else: endindex = plane_endindex(verweise, ergebnisse, quelle=quelle, vault=konfiguration.vault) aufgeloest = haenge_offene_an(ergebnisse, endindex) # Bericht nach Pfad gruppiert; Zeilen der Verweisrunde stehen unter den Zeilen ihrer Datei. gruppen: dict[Path, list[Ergebnis]] = {} for ergebnis in ergebnisse: gruppen.setdefault(ergebnis.pfad, []).append(ergebnis) for pfad, zeilen in gruppen.items(): anzeige = _anzeige(pfad, quelle) for ergebnis in zeilen: print(f" {ergebnis.aktion:<14} {anzeige}\n {ergebnis.text}") zaehler: dict[str, int] = {} for ergebnis in ergebnisse: zaehler[ergebnis.aktion] = zaehler.get(ergebnis.aktion, 0) + 1 print("\n" + ", ".join(f"{anzahl}× {name}" for name, anzahl in sorted(zaehler.items()))) if not anwenden and aufgeloest: print(f"{aufgeloest}× Verweis wird in der Verweisrunde aufgelöst") archiv_fehler = _archiviere(gruppen, quelle, anwenden=anwenden) if not anwenden: print("Probelauf — nichts geschrieben. Mit --anwenden ausführen.") return 1 if zaehler.get("FEHLER", 0) or archiv_fehler else 0 def main() -> int: return fuehre_aus(sys.argv[1:]) if __name__ == "__main__": raise SystemExit(main())