included scripts for api interaction

2026-06-17 11:41:36 +02:00
parent a697e293cb
commit 3b3069ca8c
16 changed files with 573 additions and 0 deletions
@@ -0,0 +1,241 @@
+#!/usr/bin/env python3
+"""Gemeinsame Funktionen für die Wegwichtel-API-Werkzeuge."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import getpass
+import json
+import os
+import sys
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable
+from urllib.error import HTTPError, URLError
+from urllib.parse import urljoin
+from urllib.request import Request, urlopen
+
+DEFAULT_BASE_URL = "https://wegwichtel.de"
+
+
+class ApiError(RuntimeError):
+    def __init__(self, status: int, message: str, payload: Any = None):
+        super().__init__(f"HTTP {status}: {message}")
+        self.status = status
+        self.payload = payload
+
+
+@dataclass
+class Credentials:
+    username: str
+    password: str
+
+
+class ApiClient:
+    """Kleine HTTP-Schicht mit dynamischer Basic-Auth-Nachfrage bei 401."""
+
+    def __init__(self, base_url: str, username: str | None = None, password: str | None = None):
+        self.base_url = base_url.rstrip("/") + "/"
+        self.credentials = Credentials(username, password) if username and password else None
+
+    def prompt_credentials(self) -> None:
+        default_user = self.credentials.username if self.credentials else os.getenv("WEGWICHTEL_USER", "")
+        username = input(f"Benutzername{f' [{default_user}]' if default_user else ''}: ").strip() or default_user
+        if not username:
+            raise ApiError(401, "Für den geschützten API-Zugriff ist ein Benutzername erforderlich.")
+        password = getpass.getpass("Passwort: ")
+        self.credentials = Credentials(username, password)
+
+    def _authorization_header(self) -> str | None:
+        if not self.credentials:
+            return None
+        token = base64.b64encode(
+            f"{self.credentials.username}:{self.credentials.password}".encode("utf-8")
+        ).decode("ascii")
+        return f"Basic {token}"
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        json_body: dict[str, Any] | None = None,
+        fields: dict[str, Any] | None = None,
+        files: dict[str, Path] | None = None,
+        expected: tuple[int, ...] = (200, 201),
+    ) -> Any:
+        data: bytes | None = None
+        headers = {"Accept": "application/json"}
+
+        if files:
+            data, content_type = encode_multipart(fields or {}, files)
+            headers["Content-Type"] = content_type
+        elif json_body is not None:
+            data = json.dumps(json_body, ensure_ascii=False).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+        elif fields:
+            data = json.dumps(fields, ensure_ascii=False).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+
+        url = urljoin(self.base_url, path.lstrip("/"))
+        attempts = 0
+        while True:
+            request_headers = dict(headers)
+            authorization = self._authorization_header()
+            if authorization:
+                request_headers["Authorization"] = authorization
+
+            request = Request(url, data=data, headers=request_headers, method=method.upper())
+            try:
+                with urlopen(request, timeout=120) as response:
+                    raw = response.read()
+                    status = response.status
+                    if status not in expected:
+                        raise ApiError(status, decode_error(raw), parse_json(raw))
+                    return parse_response(raw, response.headers.get("Content-Type", ""))
+            except HTTPError as error:
+                raw = error.read()
+                if error.code == 401 and attempts < 3:
+                    print("Die API verlangt Zugangsdaten (HTTP 401).", file=sys.stderr)
+                    self.prompt_credentials()
+                    attempts += 1
+                    continue
+                raise ApiError(error.code, decode_error(raw), parse_json(raw)) from error
+            except URLError as error:
+                raise RuntimeError(f"API nicht erreichbar: {error.reason}") from error
+
+
+def encode_multipart(fields: dict[str, Any], files: dict[str, Path]) -> tuple[bytes, str]:
+    """Erzeugt multipart/form-data; Dateitypen werden absichtlich nicht mitgesendet."""
+    boundary = f"----wegwichtel-{uuid.uuid4().hex}"
+    chunks: list[bytes] = []
+
+    for name, value in fields.items():
+        if value is None:
+            continue
+        chunks.extend([
+            f"--{boundary}\r\n".encode(),
+            f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
+            str(value).encode("utf-8"),
+            b"\r\n",
+        ])
+
+    for name, file_path in files.items():
+        path = Path(file_path).expanduser().resolve()
+        if not path.is_file():
+            raise FileNotFoundError(f"Datei nicht gefunden: {path}")
+        safe_name = path.name.replace('"', "_")
+        chunks.extend([
+            f"--{boundary}\r\n".encode(),
+            f'Content-Disposition: form-data; name="{name}"; filename="{safe_name}"\r\n\r\n'.encode(),
+            path.read_bytes(),
+            b"\r\n",
+        ])
+
+    chunks.append(f"--{boundary}--\r\n".encode())
+    return b"".join(chunks), f"multipart/form-data; boundary={boundary}"
+
+
+def parse_json(raw: bytes) -> Any:
+    if not raw:
+        return None
+    try:
+        return json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None
+
+
+def parse_response(raw: bytes, content_type: str) -> Any:
+    if not raw:
+        return None
+    if "json" in content_type.lower():
+        parsed = parse_json(raw)
+        if parsed is not None:
+            return parsed
+    return raw.decode("utf-8", errors="replace")
+
+
+def decode_error(raw: bytes) -> str:
+    parsed = parse_json(raw)
+    if isinstance(parsed, dict):
+        return str(parsed.get("message") or parsed.get("error") or parsed)
+    text = raw.decode("utf-8", errors="replace").strip()
+    return text or "Unbekannter API-Fehler"
+
+
+def add_common_arguments(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--base-url", help="API-Basisadresse, Standard: WEGWICHTEL_URL oder https://wegwichtel.de")
+    parser.add_argument("--username", help="Benutzername; Passwort wird bei HTTP 401 sicher abgefragt")
+
+
+def client_from_args(args: argparse.Namespace) -> ApiClient:
+    base_url = args.base_url or os.getenv("WEGWICHTEL_URL") or DEFAULT_BASE_URL
+    username = args.username or os.getenv("WEGWICHTEL_USER")
+    password = os.getenv("WEGWICHTEL_PASSWORD") if username else None
+    return ApiClient(base_url, username, password)
+
+
+def ask(label: str, value: Any = None, *, required: bool = False, cast: Callable[[str], Any] = str) -> Any:
+    if value is not None:
+        return cast(str(value))
+    while True:
+        raw = input(f"{label}: ").strip()
+        if raw:
+            try:
+                return cast(raw)
+            except (TypeError, ValueError) as error:
+                print(f"Ungültiger Wert: {error}", file=sys.stderr)
+                continue
+        if not required:
+            return None
+        print("Dieses Feld ist erforderlich.", file=sys.stderr)
+
+
+def ask_with_default(label: str, current: Any, cast: Callable[[str], Any] = str) -> Any:
+    shown = "" if current is None else str(current)
+    while True:
+        raw = input(f"{label} [{shown}]: ").strip()
+        if not raw:
+            return current
+        try:
+            return cast(raw)
+        except (TypeError, ValueError) as error:
+            print(f"Ungültiger Wert: {error}", file=sys.stderr)
+
+
+def ask_path(label: str, value: str | None = None, *, required: bool = True) -> Path | None:
+    if value is not None:
+        path = Path(value).expanduser().resolve()
+        if not path.is_file():
+            raise FileNotFoundError(f"Datei nicht gefunden: {path}")
+        return path
+
+    while True:
+        raw = input(f"{label}: ").strip()
+        if not raw and not required:
+            return None
+        if not raw:
+            print("Dieses Feld ist erforderlich.", file=sys.stderr)
+            continue
+        path = Path(raw).expanduser().resolve()
+        if path.is_file():
+            return path
+        print(f"Datei nicht gefunden: {path}", file=sys.stderr)
+
+
+def confirm(question: str) -> bool:
+    return input(f"{question} [j/N]: ").strip().lower() in {"j", "ja", "y", "yes"}
+
+
+def print_result(result: Any) -> None:
+    print(json.dumps(result, ensure_ascii=False, indent=2) if not isinstance(result, str) else result)
+
+
+def run(action: Callable[[], Any]) -> None:
+    try:
+        print_result(action())
+    except (ApiError, RuntimeError, FileNotFoundError, ValueError) as error:
+        print(f"Fehler: {error}", file=sys.stderr)
+        raise SystemExit(1) from error