""" paperless_client.py - REST-Client für Paperless-ngx. Bewusst OHNE requests: Die in LibreOffice mitgelieferte Python-Laufzeit bringt nur die Standardbibliothek mit. Multipart-Uploads werden deshalb von Hand kodiert. """ import json import re import mimetypes import os import ssl import time import urllib.error import urllib.parse import urllib.request import uuid TASK_TIMEOUT = 900 TASK_POLL = 1.5 # Obergrenze für Antwortkörper. Ohne Begrenzung könnte eine fehlerhafte # oder böswillige Gegenstelle den Arbeitsspeicher füllen, indem sie eine # endlose Antwort liefert. MAX_JSON = 32 * 1024 * 1024 # 32 MB für Metadaten MAX_DOWNLOAD = 512 * 1024 * 1024 # 512 MB für Dokumente # Eigene Kennung statt der Vorgabe von urllib. # # urllib sendet standardmäßig "Python-urllib/3.x". Diese Kennung steht in # den Sperrlisten gängiger Schutzsysteme (CrowdSec http-bad-user-agent, # fail2ban, WAF-Regelwerke), weil sie typisch für Scanner ist. Folge: Der # Zugriff über einen Reverse Proxy scheitert mit HTTP 403, während # derselbe Aufruf direkt an die Anwendung funktioniert - ein schwer zu # findendes Fehlerbild, zumal die Meldung von einem Dienst stammen kann, # der mit dem Ziel nichts zu tun hat. # # Über den Parameter user_agent bei PaperlessClient anpassbar, falls eine # Regel auch diesen Namen abweist. USER_AGENT = "PaperlessLibreOffice/0.1 (+LibreOffice extension)" class PaperlessError(Exception): pass class _SicherRedirect(urllib.request.HTTPRedirectHandler): """ Weiterleitungen nur nach http und https zulassen. Der Vorgabe-Handler von urllib erlaubt auch ftp. Eine böswillige oder übernommene Gegenstelle könnte damit auf ein anderes Protokoll umlenken. Für eine REST-Schnittstelle gibt es dafür keinen Grund. """ def redirect_request(self, req, fp, code, msg, headers, newurl): if urllib.parse.urlparse(newurl).scheme not in ("http", "https"): raise PaperlessError( "Weiterleitung auf ein nicht zugelassenes Protokoll " "abgewiesen: %s" % newurl[:120]) return super().redirect_request(req, fp, code, msg, headers, newurl) class PaperlessClient: def __init__(self, base_url, token, verify_tls=True, timeout=60, user_agent=None): # Nur http und https zulassen. Ohne Prüfung liesse sich hier # file:// oder ftp:// hinterlegen, und urllib würde das bedienen - # aus einer Adresseingabe würde ein Dateizugriff. parsed = urllib.parse.urlparse((base_url or "").strip()) if parsed.scheme not in ("http", "https") or not parsed.netloc: raise PaperlessError( "Ungültige Basisadresse. Erwartet wird http:// oder " "https:// mit Rechnernamen, z. B. https://dms.example.org") self.base = base_url.strip().rstrip("/") self.token = (token or "").strip() self.timeout = timeout self.user_agent = user_agent or USER_AGENT self._opener = urllib.request.build_opener(_SicherRedirect()) self._ctx = None if not verify_tls: # Nur für Testinstanzen mit selbstsigniertem Zertifikat. self._ctx = ssl.create_default_context() self._ctx.check_hostname = False self._ctx.verify_mode = ssl.CERT_NONE # ------------------------------------------------------------ intern def _open(self, req): req.add_header("Authorization", "Token %s" % self.token) req.add_header("User-Agent", self.user_agent) req.add_header("Accept", "application/json, */*") try: return self._opener.open(req, timeout=self.timeout) \ if self._ctx is None else \ urllib.request.urlopen(req, timeout=self.timeout, context=self._ctx) except urllib.error.HTTPError as exc: body = "" try: body = exc.read().decode("utf-8", "replace")[:400] except Exception: pass # Die Adresse kann Suchbegriffe enthalten, aber niemals den # Token - der steht im Kopfzeilenfeld. Trotzdem vorsichtshalber # kürzen, damit lange Abfragen die Meldung nicht sprengen. raise PaperlessError("HTTP %s bei %s: %s" % (exc.code, req.full_url[:160], body)) from exc except urllib.error.URLError as exc: raise PaperlessError("Verbindung fehlgeschlagen: %s" % exc.reason) from exc def _json(self, path, params=None, method="GET", payload=None): url = "%s/api/%s" % (self.base, path.lstrip("/")) if params: url += "?" + urllib.parse.urlencode(params, doseq=True) data = None req = urllib.request.Request(url, method=method) if payload is not None: data = json.dumps(payload).encode("utf-8") req.add_header("Content-Type", "application/json") req.data = data with self._open(req) as r: raw = r.read(MAX_JSON + 1) if len(raw) > MAX_JSON: raise PaperlessError("Antwort überschreitet %d MB" % (MAX_JSON // 1024 // 1024)) return json.loads(raw.decode("utf-8")) if raw else None def _multipart(self, path, fields, filename, filedata): """Multipart/form-data von Hand, weil requests nicht verfügbar ist.""" boundary = "----paperless%s" % uuid.uuid4().hex out = [] for key, value in fields.items(): if value is None: continue out.append(("--%s\r\n" "Content-Disposition: form-data; name=\"%s\"\r\n\r\n" "%s\r\n" % (boundary, key, value)).encode("utf-8")) ctype = mimetypes.guess_type(filename)[0] or "application/octet-stream" out.append(("--%s\r\n" "Content-Disposition: form-data; name=\"document\"; " "filename=\"%s\"\r\n" "Content-Type: %s\r\n\r\n" % (boundary, os.path.basename(filename), ctype)).encode("utf-8")) out.append(filedata) out.append(("\r\n--%s--\r\n" % boundary).encode("utf-8")) body = b"".join(out) url = "%s/api/%s" % (self.base, path.lstrip("/")) req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "multipart/form-data; boundary=%s" % boundary) with self._open(req) as r: raw = r.read().decode("utf-8") try: return json.loads(raw) except ValueError: return raw.strip().strip('"') # ------------------------------------------------------------ Aufgaben def wait_for_task(self, task_id, progress=None): """ Wartet auf einen Konsumvorgang und liefert die Dokument-ID. Die Antwortstruktur hat sich zwischen Versionen mehrfach geändert: Status kommt klein- oder grossgeschrieben, die Dokument-ID steht mal verschachtelt in result_data, mal als Liste in related_document_ids. """ deadline = time.time() + TASK_TIMEOUT waited = 0 while time.time() < deadline: task = self._find_task(task_id) if task: status = str(task.get("status") or "").upper() if status in ("SUCCESS", "SUCCEEDED"): doc = self._doc_id(task) if doc: return doc raise PaperlessError("Aufgabe erfolgreich, aber ohne " "Dokument-ID: %s" % task) if status in ("FAILURE", "FAILED", "REVOKED"): raise PaperlessError(self._error_text(task, status)) time.sleep(TASK_POLL) waited += TASK_POLL if progress: progress(waited) raise PaperlessError("Zeitüberschreitung nach %ds (Aufgabe %s)" % (TASK_TIMEOUT, task_id)) def _find_task(self, task_id): for params in ({"task_id": task_id}, None): try: data = self._json("tasks/", params) except PaperlessError: continue rows = data if isinstance(data, list) else (data or {}).get("results", []) for t in rows: if str(t.get("task_id")) == str(task_id): return t return None @staticmethod def _doc_id(task): def as_int(v): try: return int(v) except (TypeError, ValueError): return None for key in ("related_document_ids", "document_ids"): val = task.get(key) if isinstance(val, (list, tuple)) and val: got = as_int(val[0]) if got: return got nested = task.get("result_data") if isinstance(nested, dict): for key in ("document_id", "related_document", "id"): got = as_int(nested.get(key)) if got: return got for key in ("related_document", "document_id", "related_document_id"): got = as_int(task.get(key)) if got: return got return None @staticmethod def _error_text(task, status): rd = task.get("result_data") or {} for src in (task.get("result"), rd.get("error"), rd.get("exc_message")): if src: txt = " ".join(str(src).split()) if txt and txt.lower() not in ("failure", "failed"): return txt[:400] return "Aufgabe %s ohne Meldung" % status # ------------------------------------------------------------ Dokumente def ping(self): self._json("documents/", {"page_size": 1}) return True # Felder, die die Trefferliste braucht. Ohne diese Einschränkung # liefert Paperless je Treffer auch den vollständigen Inhalt - bei # fünfzig Treffern sind das schnell mehrere Megabyte. SUCHFELDER = ("id,title,created,added,archive_serial_number," "correspondent,document_type,tags,original_file_name") # Zeichen mit besonderer Bedeutung in der Abfragesprache des # Suchindex. Unbehandelt führen sie zu Syntaxfehlern oder zu # Treffern, die niemand erwartet - ein Bindestrich im Namen wäre # sonst ein Ausschluss. _SONDERZEICHEN = re.compile(r'[+\-&|!(){}\[\]^"~*?:\\/]') def search(self, query, limit=50): """ Dokumente suchen. Vier Wege, in dieser Reihenfolge: 1. Archiv-Seriennummer, wenn die Eingabe nur aus Ziffern besteht. Sie ist ein eigenes Feld - die Volltextsuche findet sie NICHT. 2. Die GLOBALE SUCHE von Paperless (/api/search/). Das ist dieselbe Schnittstelle, die die Suchleiste oben in der Weboberfläche bedient. Sie verbindet Titelsuche und Volltext und liefert damit die Ergebnisse, die man aus der Oberfläche kennt. 3. Teilzeichenkette in Titel und Inhalt. Greift, wenn die globale Suche nichts liefert oder in der eingesetzten Version fehlt. 4. Volltextsuche über den Index, mit UND-Verknüpfung. Sie beherrscht Wortstämme und findet auch Beugungsformen. Zwei Fallstricke, die hier berücksichtigt sind: * Bei der Volltextsuche darf KEIN ordering mitgegeben werden. Paperless sortiert dort nach Relevanz; eine eigene Sortierangabe führte dazu, dass die Suchanfrage verworfen wurde und schlicht die zuletzt hinzugefügten Dokumente kamen. * Ohne UND-Verknüpfung sucht der Index mit ODER. "Anschreiben Markus Straub" lieferte dann alles, was irgendeines der drei Wörter enthält - der gesuchte Treffer ging darin unter. """ q = (query or "").strip() if not q: return self._suchlauf({"ordering": "-added"}, limit) # 1. Seriennummer if q.isdigit(): treffer = self._suchlauf({"archive_serial_number": q}, limit) if treffer: return treffer # 2. Globale Suche wie in der Weboberfläche treffer = self._globale_suche(q, limit) if treffer: return treffer # 3. Titel und Inhalt treffer = self._suchlauf({"title_content": q, "ordering": "-created"}, limit) if treffer: return treffer # 4. Volltext mit UND, ohne ordering. # # Nach dem Ersetzen der Sonderzeichen ERNEUT zerlegen: Aus # "Müller-Lüdenscheidt" würde sonst ein Wort mit Leerzeichen, # und die UND-Verknüpfung wäre wieder aufgehoben. worte = self._SONDERZEICHEN.sub(" ", q).split() if not worte: return [] return self._suchlauf({"query": " AND ".join(worte)}, limit) def _globale_suche(self, q, limit): """ Die Schnittstelle hinter der Suchleiste der Weboberfläche. Sie gibt es erst ab neueren Paperless-Ständen, und die Form der Antwort ist nicht überall gleich - mal ein Feld "documents", mal "results". Deshalb wird beides angenommen und bei jedem Fehler stillschweigend auf den nächsten Weg ausgewichen. """ try: antwort = self._json("search/", {"query": q, "page_size": limit}) except PaperlessError: return [] if not antwort: return [] for schluessel in ("documents", "results"): rows = antwort.get(schluessel) if isinstance(rows, list) and rows: # Die globale Suche liefert je nach Version verschiedene # Objektarten. Nur echte Dokumente durchlassen. return [r for r in rows if isinstance(r, dict) and "id" in r and "title" in r][:limit] return [] def _suchlauf(self, zusatz, limit): params = {"page_size": limit, "fields": self.SUCHFELDER} params.update(zusatz) try: return (self._json("documents/", params) or {}).get("results", []) except PaperlessError: # Ein nicht unterstützter Parameter soll nicht den ganzen # Dialog zum Scheitern bringen - der nächste Weg wird # ohnehin probiert. return [] def document(self, doc_id, full_perms=False): params = {"full_perms": "true"} if full_perms else None return self._json("documents/%d/" % int(doc_id), params) def version_details(self, doc): """ Reichert die versions-Liste eines Dokuments um Format und Dateiname an. Die eingebettete Liste enthält nur id, added, version_label, checksum und is_root - nicht aber den Dateityp. Jede Version ist aber selbst ein Dokumentdatensatz, also je Version ein Abruf. Vertretbar, weil das nur beim bewussten Öffnen eines Dokuments passiert. Rückgabe absteigend nach Versionsnummer, also neueste zuerst - so wie Paperless die Liste selbst liefert. """ versions = doc.get("versions") or [] if not versions: # Dokument ohne Versionskette: es selbst ist die einzige Fassung. return [{ "id": doc["id"], "added": doc.get("added"), "version_label": None, "is_root": True, "no": 1, "mime_type": doc.get("mime_type"), "filename": doc.get("original_file_name") or "", }] # Paperless liefert absteigend. Die Versionsnummer ergibt sich aus # der Position von unten gezählt. total = len(versions) out = [] for idx, v in enumerate(versions): entry = dict(v) entry["no"] = total - idx try: full = self.document(v["id"]) entry["mime_type"] = full.get("mime_type") entry["filename"] = full.get("original_file_name") or "" except PaperlessError: entry["mime_type"] = None entry["filename"] = "" out.append(entry) return out def download(self, doc_id, original=True): """Laedt die Datei. original=True liefert das unveraenderte Original.""" url = "%s/api/documents/%d/download/" % (self.base, int(doc_id)) if original: url += "?original=true" req = urllib.request.Request(req_url := url) with self._open(req) as r: # Der Dateiname stammt aus einem Kopfzeilenfeld der Gegenstelle # und ist damit nicht vertrauenswürdig. Er wird hier nur roh # weitergereicht; die Bereinigung passiert beim Ablegen. disp = r.headers.get("Content-Disposition", "") name = None if "filename=" in disp: name = disp.split("filename=", 1)[1].strip().strip('";') name = urllib.parse.unquote(name) data = r.read(MAX_DOWNLOAD + 1) if len(data) > MAX_DOWNLOAD: raise PaperlessError( "Die Datei überschreitet die Obergrenze von %d MB." % (MAX_DOWNLOAD // 1024 // 1024)) return data, name def update_version(self, doc_id, filename, filedata, version_label=None): """Haengt eine neue Version an. Liefert die Aufgaben-Kennung.""" return self._multipart("documents/%d/update_version/" % int(doc_id), {"version_label": version_label}, filename, filedata) def post_document(self, filename, filedata, fields=None): """ Legt ein NEUES Dokument an (eigene Versionskette). Anders als update_version, das eine weitere Fassung an ein bestehendes Dokument hängt. Liefert die Aufgaben-Kennung. """ data = {} for key, value in (fields or {}).items(): if value in (None, "", []): continue data[key] = value return self._multipart("documents/post_document/", data, filename, filedata) def patch_document(self, doc_id, payload): return self._json("documents/%d/" % int(doc_id), method="PATCH", payload=payload) # ------------------------------------------------------------ Stammdaten def _all(self, path, fields="id,name"): out, page = [], 1 while True: data = self._json(path, {"page_size": 200, "page": page, "fields": fields}) if not data: break out.extend(data.get("results", [])) if not data.get("next"): break page += 1 return out def tags(self): return self._all("tags/") def correspondents(self): return self._all("correspondents/") def document_types(self): return self._all("document_types/") def storage_paths(self): return self._all("storage_paths/") def custom_fields(self): return self._all("custom_fields/", fields="id,name,data_type") def create(self, path, name): return self._json(path, method="POST", payload={"name": name}) # ------------------------------------------------------------ Freigabe def share_link(self, doc_id, days=7, file_version="original"): """ Erzeugt einen zeitlich begrenzten, unauthentifizierten Link. Der Endpunkt heißt je nach Version share_links/ oder sharelinks/. Beide werden probiert. """ from datetime import datetime, timedelta, timezone expiration = None if days: expiration = (datetime.now(timezone.utc) + timedelta(days=int(days))).isoformat() payload = {"document": int(doc_id), "file_version": file_version} if expiration: payload["expiration"] = expiration last = None for path in ("share_links/", "sharelinks/"): try: res = self._json(path, method="POST", payload=payload) slug = res.get("slug") if isinstance(res, dict) else None if slug: return "%s/share/%s" % (self.base, slug), res return None, res except PaperlessError as exc: last = exc raise last