"""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, timeout: int | None = None): self.ensure_auth() url = f"{self.base}{path}" r = self.session.get(url, params=params, timeout=timeout or 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, timeout=timeout) 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 get_ticket(self, ticket_id: int) -> dict: return self._get(f"/api/v1/tickets/{ticket_id}", params={"expand": "true"}) def ticket_articles(self, ticket_id: int, timeout: int | None = None) -> list[dict]: return self._get(f"/api/v1/ticket_articles/by_ticket/{ticket_id}", timeout=timeout) def get_attachment(self, ticket_id: int, article_id: int, att_id: int, timeout: int = 30) -> bytes: """Rohbytes eines Anhangs herunterladen.""" self.ensure_auth() url = (f"{self.base}/api/v1/ticket_attachment/" f"{ticket_id}/{article_id}/{att_id}") r = self.session.get(url, timeout=timeout) if r.status_code in (401, 403) and not config.ZAMMAD_TOKEN: self._authed = False self._login_session() r = self.session.get(url, timeout=timeout) if r.status_code != 200: raise ZammadError(f"Anhang {att_id} -> HTTP {r.status_code}") return r.content # ----------------------------------------------------------- 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 _write(self, method: str, 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.request(method, 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._write(method, path, payload, _retry=False) if r.status_code not in (200, 201): raise ZammadError(f"{method} {path} -> HTTP {r.status_code}: {r.text[:300]}") return r.json() def _post(self, path: str, payload: dict, _retry: bool = True): return self._write("POST", path, payload, _retry) def _put(self, path: str, payload: dict, _retry: bool = True): return self._write("PUT", path, payload, _retry) def ticket_states(self) -> list[dict]: return self._get("/api/v1/ticket_states") def set_state(self, ticket_id: int, state_id: int, pending_time: str | None = None) -> dict: """Status eines Tickets setzen (an EPI wirksam).""" payload = {"state_id": state_id} if pending_time: payload["pending_time"] = pending_time return self._put(f"/api/v1/tickets/{ticket_id}", payload) def create_article(self, ticket_id: int, body: str, attachments: list[dict] | None = None) -> dict: """Antwort/Nachricht an ein bestehendes Ticket anhängen (an EPI sichtbar).""" payload = { "ticket_id": ticket_id, "body": body, "content_type": "text/plain", "type": "web", "internal": False, } if attachments: payload["attachments"] = attachments return self._post("/api/v1/ticket_articles", payload) def create_ticket(self, title: str, group: str, body: str, attachments: list[dict] | None = None) -> dict: """Neues Ticket beim EPI-Support anlegen.""" article = { "subject": title, "body": body, "content_type": "text/plain", "type": "web", "internal": False, } if attachments: article["attachments"] = attachments return self._post("/api/v1/tickets", { "title": title, "group": group, "article": article, })