epi/EpiSupport-Zammad-Scraper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add EPI-Support monitoring & analytics platform

Browse Source 
Internal Flask app that monitors the epi-helpdesk.zammad.com tickets via
the Zammad API, deployable as a systemd service behind Apache under
/episupport.

Features:
- Poller: session-login API sync, snapshot diff to detect agent changes
  (status, owner, priority, closures, new replies); email notifications
- Dashboard: ticket overview with sorting, "not closed" filter and
  highlighting of tickets with unconfirmed changes; per-item and bulk
  "mark as read"
- Analytics: KPIs, response/resolution times, monthly SVG trend chart,
  distributions (status/priority/group/agent), Excel export
- Conversation view: full per-ticket thread cached in the DB (avoids
  proxy timeouts), inline image attachments, safe HTML-to-text rendering
- Write actions: reply to tickets and create new tickets with a
  confirmation step and attachment upload (incl. clipboard paste);
  self-triggered actions are synced back so they aren't flagged as changes
- Local timezone display (Europe/Berlin) for all timestamps

Includes deploy tooling: setup.sh, systemd units and Apache config.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
leopold.strobl
2026-07-01 09:19:53 +02:00
co-authored by Claude Opus 4.8
parent da2f7c3c86
commit 1041f283a4
17 changed files with 600 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config.py
+2
View File
@@ -31,6 +31,8 @@ POLL_INTERVAL_SECONDS = int(os.getenv("POLL_INTERVAL_SECONDS", "180"))
# --- Web ---
URL_PREFIX = os.getenv("EPISUPPORT_URL_PREFIX", "/episupport").rstrip("/")
PORT = int(os.getenv("EPISUPPORT_PORT", "8071"))
# Zeitzone für die Anzeige (Zammad liefert UTC)
DISPLAY_TZ = os.getenv("DISPLAY_TZ", "Europe/Berlin")
# --- E-Mail ---
NOTIFY_ENABLED = _b(os.getenv("NOTIFY_ENABLED"), False)
app/conversation.py
+157
View File
@@ -0,0 +1,157 @@
"""Konversation (Artikel) eines Tickets: in der DB zwischenspeichern und laden.
So muss die Leseseite nicht live auf Zammad warten (vermeidet Proxy-Timeouts).
Der Poller füllt/aktualisiert die Artikel im Hintergrund; die Webseite liest
nur aus der DB (mit einmaligem Lazy-Load als Rückfall).
"""
import json
import logging
from datetime import datetime, timezone
from . import config, db, textutil
log = logging.getLogger("episupport.conversation")
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
ATT_CACHE = config.DATA_DIR / "att_cache"
ATT_CACHE.mkdir(exist_ok=True)
def _attachment_meta(article) -> list[dict]:
"""Relevante Anhang-Metadaten (id, filename, mime) je Artikel."""
out = []
for x in (article.get("attachments") or []):
fn = x.get("filename")
if not fn or fn == "message.html":
continue
prefs = x.get("preferences") or {}
mime = prefs.get("Mime-Type") or prefs.get("Content-Type") or ""
out.append({"id": x.get("id"), "filename": fn, "mime": mime})
return out
def save_articles(ticket_id: int, articles: list[dict], conn=None) -> int:
"""Bereits abgerufene Artikel in die DB schreiben (INSERT OR REPLACE).
Wird `conn` übergeben, wird diese bestehende Verbindung genutzt (wichtig
im Poller, der bereits eine Transaktion offen hält – sonst „database is
locked"). Ohne `conn` wird eine eigene, kurze Transaktion verwendet.
"""
now = _now()
def _write(c):
for a in articles:
c.execute(
"INSERT OR REPLACE INTO articles (article_id, ticket_id, sender, "
"frm, subject, body, content_type, internal, created_at, "
"attachments, fetched_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(a.get("id"), ticket_id, a.get("sender"), a.get("from"),
a.get("subject"), a.get("body"), a.get("content_type"),
1 if a.get("internal") else 0, a.get("created_at"),
json.dumps(_attachment_meta(a)), now),
)
if conn is not None:
_write(conn)
else:
with db.get_conn() as own:
_write(own)
return len(articles)
def store_articles(client, ticket_id: int, timeout: int | None = None,
conn=None) -> int:
"""Artikel live von Zammad holen und speichern."""
articles = client.ticket_articles(ticket_id, timeout=timeout)
return save_articles(ticket_id, articles, conn=conn)
def load_articles(ticket_id: int) -> list[dict]:
"""Aufbereitete Artikel aus der DB (HTML sicher zu Text bereinigt)."""
with db.get_conn() as conn:
rows = conn.execute(
"SELECT * FROM articles WHERE ticket_id=? ORDER BY created_at, article_id",
(ticket_id,)
).fetchall()
out = []
for r in rows:
body = r["body"] or ""
if (r["content_type"] or "").startswith("text/html"):
body = textutil.clean_html(body)
out.append({
"id": r["article_id"],
"sender": r["sender"],
"frm": r["frm"] or "",
"subject": r["subject"],
"created_at": r["created_at"],
"internal": r["internal"],
"body": body,
"attachments": _parse_attachments(r["attachments"]),
})
return out
def _parse_attachments(raw) -> list[dict]:
"""Verträgt altes Format (nur Dateinamen) und neues (id/filename/mime)."""
items = json.loads(raw or "[]")
out = []
for x in items:
if isinstance(x, str): # altes Cache-Format
out.append({"id": None, "filename": x, "mime": "", "is_image": False})
else:
mime = x.get("mime") or ""
out.append({
"id": x.get("id"),
"filename": x.get("filename"),
"mime": mime,
"is_image": mime.startswith("image/"),
})
return out
def find_attachment(article_id: int, att_id: int) -> dict | None:
"""Metadaten eines Anhangs (Mime/Dateiname) aus der DB – für die Auslieferung."""
with db.get_conn() as conn:
row = conn.execute(
"SELECT attachments FROM articles WHERE article_id=?", (article_id,)
).fetchone()
if not row:
return None
for a in _parse_attachments(row["attachments"]):
if a["id"] == att_id:
return a
return None
def get_attachment_bytes(client, ticket_id: int, article_id: int, att_id: int) -> bytes:
"""Anhang-Bytes liefern; einmal von Zammad holen und auf Platte cachen."""
cache = ATT_CACHE / f"{article_id}_{att_id}"
if cache.exists():
return cache.read_bytes()
data = client.get_attachment(ticket_id, article_id, att_id)
try:
cache.write_bytes(data)
except OSError:
pass
return data
def has_articles(ticket_id: int) -> bool:
with db.get_conn() as conn:
return conn.execute(
"SELECT 1 FROM articles WHERE ticket_id=? LIMIT 1", (ticket_id,)
).fetchone() is not None
def tickets_without_articles(limit: int = 10) -> list[int]:
with db.get_conn() as conn:
rows = conn.execute(
"SELECT id FROM tickets WHERE id NOT IN "
"(SELECT DISTINCT ticket_id FROM articles) "
"ORDER BY zammad_updated_at DESC LIMIT ?", (limit,)
).fetchall()
return [r["id"] for r in rows]
app/db.py
+16
View File
@@ -53,6 +53,22 @@ CREATE INDEX IF NOT EXISTS idx_changes_detected ON changes(detected_at DESC);
CREATE INDEX IF NOT EXISTS idx_changes_ticket ON changes(ticket_id);
CREATE INDEX IF NOT EXISTS idx_changes_seen ON changes(seen);
CREATE TABLE IF NOT EXISTS articles (
article_id INTEGER PRIMARY KEY,
ticket_id INTEGER NOT NULL,
sender TEXT,
frm TEXT,
subject TEXT,
body TEXT,
content_type TEXT,
internal INTEGER,
created_at TEXT,
attachments TEXT,
fetched_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_articles_ticket ON articles(ticket_id, created_at);
CREATE TABLE IF NOT EXISTS sync_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
started_at TEXT,
app/poller.py
+42 -2
View File
@@ -5,7 +5,7 @@ import logging
import time
from datetime import datetime, timezone
from . import config, db, notify
from . import config, conversation, db, notify
from .zammad import ZammadClient, ZammadError
log = logging.getLogger("episupport.poller")
@@ -111,7 +111,12 @@ def _detect_changes(conn, client, new_t: dict, old_row) -> list[dict]:
new_count = new_t.get("article_count") or 0
if new_count > old_count:
detail = None
articles = client.ticket_articles(new_t["id"])
try:
articles = client.ticket_articles(new_t["id"])
conversation.save_articles(new_t["id"], articles, conn=conn) # cachen
except ZammadError as e:
log.warning("Artikel für Ticket %s nicht abrufbar: %s", new_t["id"], e)
articles = []
if articles:
last = articles[-1]
sender = last.get("sender") # Agent / Customer / System
@@ -133,6 +138,35 @@ def _detect_changes(conn, client, new_t: dict, old_row) -> list[dict]:
return found
def sync_ticket(client: ZammadClient, ticket_id: int) -> None:
"""Ein einzelnes Ticket (+ Artikel) sofort in die DB übernehmen.
Wird nach einer EIGENEN Aktion (Antwort/Neuanlage über den Client) gerufen,
damit der nächste Poll-Durchlauf keinen Unterschied sieht und die selbst
ausgelöste Änderung weder im Feed erscheint noch eine Mail auslöst.
"""
t = client.get_ticket(ticket_id)
with db.get_conn() as conn:
existing = conn.execute(
"SELECT id FROM tickets WHERE id=?", (ticket_id,)
).fetchone()
_upsert(conn, _map_ticket(t), is_new=(existing is None))
try:
conversation.store_articles(client, ticket_id)
except ZammadError as e:
log.warning("Artikel-Sync nach eigener Aktion (Ticket %s): %s", ticket_id, e)
def _backfill_articles(client: ZammadClient, limit: int = 10) -> None:
"""Pro Durchlauf einige Tickets ohne gespeicherte Konversation nachladen,
damit die Leseansicht nach und nach komplett aus der DB bedient wird."""
for tid in conversation.tickets_without_articles(limit):
try:
conversation.store_articles(client, tid)
except ZammadError as e:
log.warning("Backfill für Ticket %s fehlgeschlagen: %s", tid, e)
def poll_once(client: ZammadClient) -> int:
"""Ein Abruf-Durchlauf. Gibt Anzahl erkannter Änderungen zurück."""
started = _now()
@@ -157,9 +191,15 @@ def poll_once(client: ZammadClient) -> int:
all_changes.append(_record_change(
conn, t, "new_ticket", "Neues Ticket", None,
f"#{t.get('number')}", None))
try:
conversation.store_articles(client, t["id"], conn=conn)
except ZammadError:
pass
else:
all_changes.extend(_detect_changes(conn, client, t, existing))
_upsert(conn, row, is_new=False)
# Backfill NACH der Haupttransaktion (eigene, kurze Verbindungen)
_backfill_articles(client)
except ZammadError as e:
status = "error"
error = str(e)
app/uploads.py
+101
View File
@@ -0,0 +1,101 @@
"""Zwischenspeicher für hochgeladene Anhänge.
Beim Bestätigungs-Schritt werden Dateien temporär unter data/uploads/<token>
abgelegt und erst beim verbindlichen Senden gelesen und (base64) an Zammad
übergeben. Danach wird der Zwischenspeicher gelöscht.
"""
import base64
import json
import re
import shutil
import uuid
from datetime import datetime, timezone
from pathlib import Path
from . import config
UPLOAD_DIR = config.DATA_DIR / "uploads"
UPLOAD_DIR.mkdir(exist_ok=True)
_TOKEN_RE = re.compile(r"^[0-9a-f]{32}$")
def _dir(token: str) -> Path | None:
if not token or not _TOKEN_RE.match(token):
return None
return UPLOAD_DIR / token
def save_pending(files) -> tuple[str | None, list[str]]:
"""Hochgeladene Dateien (werkzeug FileStorage) zwischenspeichern.
Gibt (token, Dateinamen) zurück; (None, []) wenn nichts dabei war."""
real = [f for f in files if f and getattr(f, "filename", "")]
if not real:
return None, []
token = uuid.uuid4().hex
d = UPLOAD_DIR / token
d.mkdir(parents=True, exist_ok=True)
manifest, names = [], []
for i, f in enumerate(real):
data = f.read()
if not data:
continue
(d / str(i)).write_bytes(data)
manifest.append({
"stored": str(i),
"filename": f.filename,
"mimetype": getattr(f, "mimetype", None) or "application/octet-stream",
"size": len(data),
})
names.append(f.filename)
(d / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
return token, names
def _manifest(token: str) -> list[dict]:
d = _dir(token)
if not d:
return []
mf = d / "manifest.json"
if not mf.exists():
return []
return json.loads(mf.read_text(encoding="utf-8"))
def names(token: str) -> list[str]:
return [m["filename"] for m in _manifest(token)]
def load_attachments(token: str) -> list[dict]:
"""Zammad-Inline-Format: {filename, data(base64), mime-type}."""
d = _dir(token)
out = []
for m in _manifest(token):
p = d / m["stored"]
if not p.exists():
continue
out.append({
"filename": m["filename"],
"data": base64.b64encode(p.read_bytes()).decode("ascii"),
"mime-type": m["mimetype"],
})
return out
def discard(token: str) -> None:
d = _dir(token)
if d and d.exists():
shutil.rmtree(d, ignore_errors=True)
def cleanup_stale(max_age_seconds: int = 3600) -> None:
"""Verwaiste Zwischenspeicher (abgebrochene Vorgänge) aufräumen."""
if not UPLOAD_DIR.exists():
return
now = datetime.now(timezone.utc).timestamp()
for d in UPLOAD_DIR.iterdir():
try:
if d.is_dir() and (now - d.stat().st_mtime) > max_age_seconds:
shutil.rmtree(d, ignore_errors=True)
except OSError:
pass
app/web.py
+85 -28
View File
@@ -5,15 +5,23 @@ Das SCRIPT_NAME-Middleware sorgt dafür, dass url_for() korrekte Links mit
Präfix erzeugt.
"""
import logging
from datetime import datetime, timezone
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from flask import (Flask, Response, abort, flash, redirect, render_template,
request, url_for)
from . import analytics, charts, config, db, export, textutil
from . import analytics, charts, config, conversation, db, export, poller, uploads
from .zammad import ZammadClient, ZammadError
log = logging.getLogger("episupport.web")
try:
_DISPLAY_TZ = ZoneInfo(config.DISPLAY_TZ)
except (ZoneInfoNotFoundError, ValueError):
log.warning("Zeitzone %s nicht verfügbar – zeige UTC.", config.DISPLAY_TZ)
_DISPLAY_TZ = timezone.utc
# Lazy initialisierter Zammad-Client für Live-Abrufe (Konversation).
_client = None
@@ -25,31 +33,28 @@ def _zammad():
return _client
def _load_conversation(ticket_id):
"""Holt die Artikel eines Tickets live und bereitet sie auf.
Gibt (articles, error) zurück."""
def _sync_own_action(ticket_id):
"""Nach eigener Antwort/Neuanlage das Ticket sofort in die DB übernehmen,
damit die selbst ausgelöste Änderung nicht als Änderung gemeldet wird."""
try:
raw = _zammad().ticket_articles(ticket_id)
poller.sync_ticket(_zammad(), ticket_id)
except Exception as e: # noqa: BLE001 - darf den Sendevorgang nie stören
log.warning("sync_ticket nach eigener Aktion fehlgeschlagen (%s): %s",
ticket_id, e)
def _load_conversation(ticket_id):
"""Konversation aus der DB laden. Ist noch nichts gespeichert, einmalig
live nachladen (kurzer Timeout, damit die Seite nicht hängt) und cachen.
Gibt (articles, error) zurück."""
arts = conversation.load_articles(ticket_id)
if arts:
return arts, None
try:
conversation.store_articles(_zammad(), ticket_id, timeout=20)
except ZammadError as e:
return [], str(e)
out = []
for a in raw:
body = a.get("body") or ""
if (a.get("content_type") or "").startswith("text/html"):
body = textutil.clean_html(body)
atts = [x.get("filename") for x in (a.get("attachments") or [])
if x.get("filename") and x.get("filename") != "message.html"]
out.append({
"id": a.get("id"),
"sender": a.get("sender"), # Customer / Agent / System
"frm": a.get("from") or "",
"created_at": a.get("created_at"),
"subject": a.get("subject"),
"internal": a.get("internal"),
"body": body,
"attachments": atts,
})
return out, None
return conversation.load_articles(ticket_id), None
class PrefixMiddleware:
@@ -71,15 +76,24 @@ class PrefixMiddleware:
def create_app() -> Flask:
app = Flask(__name__, template_folder="../templates", static_folder="../static")
app.secret_key = "episupport-local" # nur für Flash-Messages, keine Logins
app.config["MAX_CONTENT_LENGTH"] = 25 * 1024 * 1024 # max. 25 MB Upload
app.wsgi_app = PrefixMiddleware(app.wsgi_app, prefix=config.URL_PREFIX)
db.init_db()
# ------------------------------------------------------------- Filter
@app.template_filter("dt")
def _dt(value):
"""UTC-Zeitstempel in lokaler Zeitzone (Standard Europe/Berlin) anzeigen."""
if not value:
return "–"
return str(value).replace("T", " ")[:16]
try:
s = str(value).replace("Z", "+00:00")
d = datetime.fromisoformat(s)
if d.tzinfo is None:
d = d.replace(tzinfo=timezone.utc)
return d.astimezone(_DISPLAY_TZ).strftime("%d.%m.%Y %H:%M")
except (ValueError, TypeError):
return str(value).replace("T", " ")[:16]
@app.context_processor
def _inject():
@@ -166,6 +180,32 @@ def create_app() -> Flask:
return render_template("ticket.html", ticket=ticket, timeline=timeline,
conversation=conversation, conv_error=conv_error)
@app.route("/ticket/<int:ticket_id>/article/<int:article_id>/att/<int:att_id>")
def attachment(ticket_id, article_id, att_id):
meta = conversation.find_attachment(article_id, att_id)
if not meta or meta["id"] is None:
abort(404)
try:
data = conversation.get_attachment_bytes(
_zammad(), ticket_id, article_id, att_id)
except ZammadError:
abort(502)
return Response(
data,
mimetype=meta["mime"] or "application/octet-stream",
headers={"Content-Disposition":
f'inline; filename="{meta["filename"]}"'},
)
@app.route("/ticket/<int:ticket_id>/refresh", methods=["POST"])
def refresh_conversation(ticket_id):
try:
conversation.store_articles(_zammad(), ticket_id, timeout=25)
flash("Konversation aktualisiert.")
except ZammadError as e:
flash("Aktualisierung fehlgeschlagen: " + str(e))
return redirect(url_for("ticket_detail", ticket_id=ticket_id))
# ----------------------------------------------------- Antworten (live)
@app.route("/ticket/<int:ticket_id>/reply", methods=["POST"])
def reply_preview(ticket_id):
@@ -176,27 +216,35 @@ def create_app() -> Flask:
if not body:
flash("Bitte zuerst einen Text eingeben.")
return redirect(url_for("ticket_detail", ticket_id=ticket_id))
uploads.cleanup_stale()
token, att_names = uploads.save_pending(request.files.getlist("files"))
return render_template(
"confirm.html",
heading="Antwort an EPI senden?",
intro=f"Diese Antwort wird live an den EPI-Support zu Ticket "
f"#{ticket['number'] if ticket else ticket_id} übermittelt:",
preview=body,
attachments=att_names,
action_url=url_for("reply_send", ticket_id=ticket_id),
cancel_url=url_for("ticket_detail", ticket_id=ticket_id),
fields={"body": body},
fields={"body": body, "token": token or ""},
)
@app.route("/ticket/<int:ticket_id>/reply/send", methods=["POST"])
def reply_send(ticket_id):
body = (request.form.get("body") or "").strip()
token = request.form.get("token") or ""
if not body:
uploads.discard(token)
return redirect(url_for("ticket_detail", ticket_id=ticket_id))
try:
_zammad().create_article(ticket_id, body)
_zammad().create_article(ticket_id, body, uploads.load_attachments(token))
_sync_own_action(ticket_id)
flash("✓ Antwort wurde an EPI gesendet.")
except ZammadError as e:
flash("✗ Fehler beim Senden: " + str(e))
finally:
uploads.discard(token)
return redirect(url_for("ticket_detail", ticket_id=ticket_id))
# -------------------------------------------------- Neues Ticket (live)
@@ -219,15 +267,18 @@ def create_app() -> Flask:
if not (title and group and body):
flash("Bitte Titel, Gruppe und Nachricht ausfüllen.")
return redirect(url_for("new_ticket_form"))
uploads.cleanup_stale()
token, att_names = uploads.save_pending(request.files.getlist("files"))
return render_template(
"confirm.html",
heading="Neues Ticket bei EPI anlegen?",
intro=f"Es wird ein neues Ticket in der Gruppe „{group}“ mit dem "
f"Titel „{title}“ live beim EPI-Support angelegt:",
preview=body,
attachments=att_names,
action_url=url_for("new_ticket_send"),
cancel_url=url_for("new_ticket_form"),
fields={"title": title, "group": group, "body": body},
fields={"title": title, "group": group, "body": body, "token": token or ""},
)
@app.route("/new/send", methods=["POST"])
@@ -235,16 +286,22 @@ def create_app() -> Flask:
title = (request.form.get("title") or "").strip()
group = (request.form.get("group") or "").strip()
body = (request.form.get("body") or "").strip()
token = request.form.get("token") or ""
if not (title and group and body):
uploads.discard(token)
return redirect(url_for("new_ticket_form"))
try:
res = _zammad().create_ticket(title, group, body)
res = _zammad().create_ticket(title, group, body,
uploads.load_attachments(token))
flash(f"✓ Ticket #{res.get('number')} wurde bei EPI angelegt.")
tid = res.get("id")
if tid:
_sync_own_action(tid)
return redirect(url_for("ticket_detail", ticket_id=tid))
except ZammadError as e:
flash("✗ Fehler beim Anlegen: " + str(e))
finally:
uploads.discard(token)
return redirect(url_for("new_ticket_form"))
@app.route("/changes")
app/zammad.py
+44 -20
View File
@@ -58,16 +58,17 @@ class ZammadClient:
self._login_session()
# ------------------------------------------------------------- requests
def _get(self, path: str, params: dict | None = None, _retry: bool = True):
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=60)
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)
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()
@@ -97,12 +98,27 @@ class ZammadClient:
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 []
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:
@@ -128,26 +144,34 @@ class ZammadClient:
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:
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)."""
return self._post("/api/v1/ticket_articles", {
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) -> dict:
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": {
"subject": title,
"body": body,
"content_type": "text/plain",
"type": "web",
"internal": False,
},
"article": article,
})