claude-werkzeuge/.claude/skills/bookstack-upload/scripts/bookstack/api.py

"""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