242 lines
8.5 KiB
Python
242 lines
8.5 KiB
Python
#!/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
|