first commit

app/zammad.py

@@ -0,0 +1,153 @@
+"""Zammad-API-Client.
+
+Unterstützt zwei Authentifizierungs-Wege:
+  * Token  (Authorization: Token token=...)  -> bevorzugt, sobald ein
+    persönlicher Token vorliegt.
+  * Session-Login (Benutzer/Passwort)        -> Fallback, solange auf der
+    Instanz "API password access" deaktiviert ist.
+
+Re-Login passiert automatisch, falls eine Session abläuft (401/403).
+"""
+import logging
+
+import requests
+
+from . import config
+
+log = logging.getLogger("episupport.zammad")
+
+
+class ZammadError(RuntimeError):
+    pass
+
+
+class ZammadClient:
+    def __init__(self):
+        self.base = config.ZAMMAD_URL
+        self.session = requests.Session()
+        self.session.headers.update({"User-Agent": "episupport-monitor/1.0"})
+        self._authed = False
+
+    # ------------------------------------------------------------------ auth
+    def _login_session(self) -> None:
+        """Login per CSRF + signin, Cookies bleiben in der Session."""
+        r = self.session.get(f"{self.base}/api/v1/signshow", timeout=30)
+        csrf = r.headers.get("CSRF-TOKEN")
+        if not csrf:
+            raise ZammadError("Kein CSRF-Token von /signshow erhalten.")
+        r = self.session.post(
+            f"{self.base}/api/v1/signin",
+            json={
+                "username": config.ZAMMAD_USER,
+                "password": config.ZAMMAD_PASSWORD,
+            },
+            headers={"X-CSRF-Token": csrf},
+            timeout=30,
+        )
+        if r.status_code != 201 and r.status_code != 200:
+            raise ZammadError(f"Login fehlgeschlagen (HTTP {r.status_code}): {r.text[:200]}")
+        self._authed = True
+        log.info("Zammad-Session aufgebaut.")
+
+    def ensure_auth(self) -> None:
+        if config.ZAMMAD_TOKEN:
+            self.session.headers["Authorization"] = f"Token token={config.ZAMMAD_TOKEN}"
+            self._authed = True
+            return
+        if not self._authed:
+            self._login_session()
+
+    # ------------------------------------------------------------- requests
+    def _get(self, path: str, params: dict | None = None, _retry: bool = True):
+        self.ensure_auth()
+        url = f"{self.base}{path}"
+        r = self.session.get(url, params=params, timeout=60)
+        if r.status_code in (401, 403) and _retry and not config.ZAMMAD_TOKEN:
+            # Session evtl. abgelaufen -> einmal neu einloggen
+            log.warning("Auth abgelaufen (HTTP %s), erneuter Login.", r.status_code)
+            self._authed = False
+            self._login_session()
+            return self._get(path, params=params, _retry=False)
+        if r.status_code != 200:
+            raise ZammadError(f"GET {path} -> HTTP {r.status_code}: {r.text[:200]}")
+        return r.json()
+
+    # ------------------------------------------------------------- endpoints
+    def me(self) -> dict:
+        return self._get("/api/v1/users/me")
+
+    def list_tickets(self) -> list[dict]:
+        """Alle für den Account sichtbaren Tickets, über alle Seiten."""
+        out: list[dict] = []
+        page = 1
+        per_page = 100
+        while True:
+            batch = self._get(
+                "/api/v1/tickets",
+                params={"expand": "true", "page": page, "per_page": per_page},
+            )
+            if not isinstance(batch, list):
+                raise ZammadError("Unerwartete Antwort bei /tickets (kein Array).")
+            out.extend(batch)
+            if len(batch) < per_page:
+                break
+            page += 1
+            if page > 100:  # Sicherheitsanker gegen Endlosschleife
+                log.warning("Pagination-Limit erreicht.")
+                break
+        return out
+
+    def ticket_articles(self, ticket_id: int) -> list[dict]:
+        try:
+            return self._get(f"/api/v1/ticket_articles/by_ticket/{ticket_id}")
+        except ZammadError as e:
+            log.warning("Artikel für Ticket %s nicht abrufbar: %s", ticket_id, e)
+            return []
+
+    # ----------------------------------------------------------- schreibend
+    def _csrf(self) -> str | None:
+        """Frischen CSRF-Token aus der Session holen (für POST per Session)."""
+        r = self.session.get(f"{self.base}/api/v1/signshow", timeout=30)
+        return r.headers.get("CSRF-TOKEN")
+
+    def _post(self, path: str, payload: dict, _retry: bool = True):
+        self.ensure_auth()
+        headers = {}
+        if not config.ZAMMAD_TOKEN:
+            csrf = self._csrf()
+            if csrf:
+                headers["X-CSRF-Token"] = csrf
+        r = self.session.post(f"{self.base}{path}", json=payload,
+                              headers=headers, timeout=60)
+        if r.status_code in (401, 403) and _retry and not config.ZAMMAD_TOKEN:
+            log.warning("Auth/CSRF abgelaufen (HTTP %s), erneuter Login.", r.status_code)
+            self._authed = False
+            self._login_session()
+            return self._post(path, payload, _retry=False)
+        if r.status_code not in (200, 201):
+            raise ZammadError(f"POST {path} -> HTTP {r.status_code}: {r.text[:300]}")
+        return r.json()
+
+    def create_article(self, ticket_id: int, body: str) -> dict:
+        """Antwort/Nachricht an ein bestehendes Ticket anhängen (an EPI sichtbar)."""
+        return self._post("/api/v1/ticket_articles", {
+            "ticket_id": ticket_id,
+            "body": body,
+            "content_type": "text/plain",
+            "type": "web",
+            "internal": False,
+        })
+
+    def create_ticket(self, title: str, group: str, body: str) -> dict:
+        """Neues Ticket beim EPI-Support anlegen."""
+        return self._post("/api/v1/tickets", {
+            "title": title,
+            "group": group,
+            "article": {
+                "subject": title,
+                "body": body,
+                "content_type": "text/plain",
+                "type": "web",
+                "internal": False,
+            },
+        })