""" paperless_addon.py - Einstiegspunkt der LibreOffice-Erweiterung. Registriert einen Protokoll-Handler für die URLs paperless.Command:*. Die Symbolleiste in Addons.xcu ruft diese URLs auf. Warum kein Eingriff in den normalen Speichern-Knopf: Paperless als echtes Speicherziel in die Datei-Dialoge einzuhaengen würde einen UCB Content Provider erfordern - C++-UNO-Arbeit in einer sehr dünn dokumentierten Ecke. Eigene Schaltflaechen sind außerdem die ehrlichere Lösung, weil sichtbar bleibt, wann eine neue Version im Archiv entsteht. """ import os import sys import traceback import uno import unohelper from com.sun.star.frame import XDispatchProvider, XDispatch from com.sun.star.lang import XInitialization, XServiceInfo from com.sun.star.beans import PropertyValue import paperless_config as cfg import paperless_dialogs as dlg import paperless_formats as fmt from paperless_i18n import t from paperless_client import PaperlessClient, PaperlessError IMPL_NAME = "de.mmorath.paperless.AddonImpl" SERVICE_NAME = "com.sun.star.frame.ProtocolHandler" PROTOCOL = "paperless.Command:" # Ladeeinstellungen für heruntergeladene Dateien. # # SICHERHEITSRELEVANT: Ein Dokument aus dem Archiv ist fremder Inhalt - # auch dann, wenn es aus dem eigenen Bestand stammt, denn dort landen # Dateien aus E-Mail-Anhängen, Scannerordnern und Freigaben. Ein ODT oder # DOCX kann Makros enthalten, die beim Öffnen ausgeführt werden, und # Verweise auf externe Inhalte, die beim Öffnen nachgeladen werden. # # MacroExecMode 1 = NEVER_EXECUTE: Makros werden nicht ausgeführt, auch # nicht nach Rückfrage. Der Anwender kann sie nicht aus # Versehen freigeben. # UpdateDocMode 0 = NO_UPDATE: keine Aktualisierung verknüpfter Inhalte, # also kein Nachladen externer Adressen beim Öffnen. MACRO_NEVER_EXECUTE = 1 UPDATE_NO_UPDATE = 0 def _prop(name, value): p = PropertyValue() p.Name, p.Value = name, value return p class PaperlessAddon(unohelper.Base, XDispatchProvider, XDispatch, XInitialization, XServiceInfo): def __init__(self, ctx): self.ctx = ctx self.frame = None self._desktop = None # ------------------------------------------------------ UNO-Gerüst def initialize(self, args): if args: self.frame = args[0] def queryDispatch(self, url, _target, _flags): return self if url.Protocol == PROTOCOL else None def queryDispatches(self, requests): return tuple(self.queryDispatch(r.FeatureURL, r.FrameName, r.SearchFlags) for r in requests) def addStatusListener(self, listener, url): pass def removeStatusListener(self, listener, url): pass def getImplementationName(self): return IMPL_NAME def supportsService(self, name): return name == SERVICE_NAME def getSupportedServiceNames(self): return (SERVICE_NAME,) # ---------------------------------------------------------- Dispatch def dispatch(self, url, _args): command = url.Path handler = { "Open": self.cmd_open, "SaveVersion": self.cmd_save_version, "SaveAs": self.cmd_save_as, "Classify": self.cmd_classify, "Share": self.cmd_share, "Settings": self.cmd_settings, "Info": self.cmd_info, }.get(command) if handler is None: return try: handler() except PaperlessError as exc: dlg.error(self.ctx, self.frame, str(exc)) except Exception as exc: # noqa: BLE001 # Der Ablaufbericht steht auf der Konsole, nicht im Dialog. Er # kann Pfade und Werte enthalten, die niemanden etwas angehen, # der zufällig auf den Bildschirm sieht. sys.stderr.write("[paperless] %s\n%s\n" % (command, traceback.format_exc())) dlg.error(self.ctx, self.frame, "Unerwarteter Fehler bei „%s“:\n\n%s\n\n" "Einzelheiten stehen auf der Konsole " "(LibreOffice aus dem Terminal starten)." % (command, str(exc)[:300])) # ---------------------------------------------------------- Helfer def _desktop_obj(self): if self._desktop is None: self._desktop = self.ctx.ServiceManager.createInstanceWithContext( "com.sun.star.frame.Desktop", self.ctx) return self._desktop def _client(self, require=True): s = cfg.load_settings() if not cfg.settings_complete(s): if require: dlg.message(self.ctx, self.frame, "Bitte zuerst Basisadresse und Token hinterlegen.") self.cmd_settings() s = cfg.load_settings() if not cfg.settings_complete(s): return None, s return PaperlessClient(s["base_url"], s["token"], s.get("verify_tls", True), user_agent=s.get("user_agent") or None), s def _current_doc(self): try: return self._desktop_obj().getCurrentComponent() except Exception: return None def _current_doc_id(self, quiet=False): doc = self._current_doc() if doc is None: if not quiet: dlg.message(self.ctx, self.frame, "Kein Dokument geöffnet.") return None, None doc_id = cfg.get_document_id(doc) if doc_id is None and not quiet: dlg.message(self.ctx, self.frame, "Dieses Dokument stammt nicht aus Paperless.\n\n" "Es wurde keine Paperless-Kennung gefunden. Öffne es " "über „Aus Paperless öffnen“, damit die " "Zuordnung erhalten bleibt.") return doc, doc_id # ---------------------------------------------------- Befehl: Öffnen def cmd_open(self): client, settings = self._client() if client is None: return picked = dlg.open_dialog(self.ctx, self.frame, client) if not picked: return full, version, readonly = picked root_id = full["id"] # Heruntergeladen wird die gewählte Version. Jede Version ist ein # eigener Dokumentdatensatz mit eigener Kennung. data, remote_name = client.download(version["id"], original=True) # Der Name stammt aus einer Kopfzeile der Gegenstelle und ist damit # nicht vertrauenswürdig. Ein Wert wie "../../.bashrc" darf nicht # aus dem Arbeitsverzeichnis herausführen - deshalb vollständige # Bereinigung, plattformunabhängig für beide Trennzeichen. name = cfg.safe_filename( remote_name or version.get("filename") or full.get("original_file_name") or ("paperless-%d.bin" % root_id), vorgabe="paperless-%d.bin" % root_id) # Ablage im Benutzerprofil statt im allgemeinen Temp-Verzeichnis. # Letzteres ist auf Mehrbenutzersystemen für alle beschreibbar; ein # anderer Benutzer könnte dort unter dem erwarteten Namen einen # symbolischen Verweis anlegen und das Schreiben umlenken. path = os.path.join(cfg.work_dir(str(root_id)), name) # Datei mit eingeschränkten Rechten anlegen. O_EXCL ist hier # bewusst nicht gesetzt, weil dieselbe Version erneut geöffnet # werden können soll - das Verzeichnis liegt aber im Benutzer- # profil und ist für andere nicht beschreibbar. fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) try: with os.fdopen(fd, "wb") as fh: fh.write(data) except Exception: # noqa: BLE001 raise url = unohelper.systemPathToFileUrl(path) doc = self._desktop_obj().loadComponentFromURL( url, "_blank", 0, (_prop("ReadOnly", readonly), _prop("MacroExecutionMode", MACRO_NEVER_EXECUTE), _prop("UpdateDocMode", UPDATE_NO_UPDATE))) # Gemerkt wird immer die WURZEL-Kennung, nicht die der geöffneten # Version. Eine neue Version wird ja an die Wurzel angehängt. if doc is not None: cfg.set_document_id(doc, root_id, settings["base_url"]) cfg.remember(url, root_id, settings["base_url"]) if readonly: dlg.message(self.ctx, self.frame, "Schreibgeschützt geöffnet.\n\nZum Ablegen einer " "neuen Version muss das Dokument bearbeitbar " "geöffnet werden.") # -------------------------------------------- Befehl: Neue Version def cmd_save_version(self): client, settings = self._client() if client is None: return doc, doc_id = self._current_doc_id() if doc_id is None: return remote = client.document(doc_id) title = remote.get("title") or str(doc_id) if not dlg.confirm( self.ctx, self.frame, "Als neue Version im Archiv ablegen?\n\n" "Dokument: %s\nPaperless-ID: %d\n\n" "Die bisherige Fassung bleibt als Version erhalten." % (title[:70], doc_id)): return path = self._store_local(doc) if path is None: return with open(path, "rb") as fh: data = fh.read() from datetime import date label = (settings.get("version_label") or "LibreOffice {datum}") \ .replace("{datum}", date.today().isoformat()) task = client.update_version(doc_id, os.path.basename(path), data, version_label=label) if not isinstance(task, str) or not task: dlg.error(self.ctx, self.frame, t("Unerwartete Antwort beim Hochladen:\n%r") % (task,)) return new_id = client.wait_for_task(task) cfg.set_document_id(doc, doc_id, settings["base_url"]) dlg.message(self.ctx, self.frame, "Neue Version abgelegt.\n\n" "Dokument: %s\nWurzel-ID: %d\nVersions-ID: %s\n" "Bezeichnung: %s" % (title[:70], doc_id, new_id, label)) def _store_local(self, doc, vorschlag=None): """ Speichert das Dokument und liefert den Pfad auf der Platte. Drei Fälle: * über „Aus Paperless öffnen“ geladen - getURL() zeigt auf die temporäre Datei, es genügt ein Speichern im Originalformat * lokal vorhandene Datei - ebenso * noch nie gespeichert (Unbenannt 1) - dann gibt es keinen URL. Statt abzubrechen wird in dasselbe temporäre Verzeichnis geschrieben, das auch beim Öffnen benutzt wird. """ url = doc.getURL() if not url: ext, filt = fmt.default_for_document(doc) name = cfg.safe_filename( (vorschlag or self._doc_title(doc)), maxlen=72) + ext path = os.path.join(cfg.work_dir("neu"), name) try: doc.storeToURL(unohelper.systemPathToFileUrl(path), (_prop("FilterName", filt), _prop("Overwrite", True))) cfg.lock_down(path, 0o600) except Exception as exc: # noqa: BLE001 dlg.error(self.ctx, self.frame, t("Das Dokument konnte nicht zwischengespeichert werden:\n%s") % exc) return None return path path = unohelper.fileUrlToSystemPath(url) # Bei unbekanntem Format gibt es keinen passenden Filter. Dann # speichert LibreOffice im eigenen Format weiter - der Anwender # sollte wissen, dass die Endung danach nicht mehr zum Inhalt passt. if not fmt.is_known(path) and doc.isModified(): if not dlg.confirm( self.ctx, self.frame, "Für dieses Format ist kein Speicherfilter hinterlegt " "(%s).\n\nLibreOffice speichert im eigenen Format. Die " "Dateiendung passt danach möglicherweise nicht mehr zum " "Inhalt.\n\nTrotzdem fortfahren?" % os.path.basename(path)): return None try: if doc.isModified(): filt = fmt.store_filter(path) if filt: doc.storeToURL(url, (_prop("FilterName", filt), _prop("Overwrite", True))) else: doc.store() except Exception as exc: # noqa: BLE001 dlg.error(self.ctx, self.frame, t("Speichern fehlgeschlagen:\n%s") % exc) return None return path @staticmethod def _doc_title(doc): """Titel aus den Dokumenteigenschaften, sonst der Fenstertitel.""" try: t = doc.getDocumentProperties().Title if t and t.strip(): return t.strip() except Exception: # noqa: BLE001 pass try: return (doc.getTitle() or "Dokument").strip() except Exception: # noqa: BLE001 return "Dokument" # ------------------------------------- Befehl: Als neues Dokument def cmd_save_as(self): """ Legt die geöffnete Datei in Paperless ab - wahlweise als neues Dokument oder als neue Version eines ANDEREN vorhandenen Dokuments. Der zweite Fall deckt eine Lücke, die Dokumentenverwaltungen gern offenlassen: Die Rechnung kommt als PDF, die bearbeitbare Fassung entsteht später. Beides gehört zusammen, aber das PDF war zuerst da - mit „Als neue Version ablegen“ allein kommt man da nicht hin, weil das immer auf das eigene Wurzeldokument zielt. """ client, settings = self._client() if client is None: return doc, doc_id = self._current_doc_id(quiet=True) if doc is None: dlg.message(self.ctx, self.frame, "Kein Dokument geöffnet.") return # Die Herkunft aus Paperless ist optional - auch eine fremde Datei # kann so ins Archiv wandern. source = {} if doc_id is not None: try: source = client.document(doc_id) except PaperlessError: source = {} if not source.get("title"): url = doc.getURL() or "" base = os.path.splitext(os.path.basename( unohelper.fileUrlToSystemPath(url)))[0] if url else "Dokument" source = {"title": base} choice = dlg.store_target_dialog(self.ctx, self.frame, client, source) if choice is None: return # Der gewählte Titel bestimmt bei einem noch nie gespeicherten # Dokument auch den Namen der temporären Datei - und damit den # Dateinamen, unter dem Paperless das Original ablegt. vorschlag = choice.get("title") if choice["mode"] == "new" else \ (choice.get("target") or {}).get("title") path = self._store_local(doc, vorschlag) if path is None: return with open(path, "rb") as fh: data = fh.read() if choice["mode"] == "existing": self._append_to_existing(client, settings, doc, choice, path, data) else: self._create_new(client, settings, doc, doc_id, source, choice, path, data) # -- Ziel: bestehendes Dokument ----------------------------------------- def _append_to_existing(self, client, settings, doc, choice, path, data): target = choice["target"] target_id = target["id"] if not dlg.confirm( self.ctx, self.frame, "Die geöffnete Datei als neue Version anhängen?\n\n" "Zieldokument: %s\nPaperless-ID: %d\n\n" "Die bisherige Fassung des Ziels bleibt als Version " "erhalten." % ((target.get("title") or "")[:70], target_id)): return from datetime import date label = (settings.get("version_label") or "LibreOffice {datum}") \ .replace("{datum}", date.today().isoformat()) task = client.update_version(target_id, os.path.basename(path), data, version_label=label) if not isinstance(task, str) or not task: dlg.error(self.ctx, self.frame, t("Unerwartete Antwort beim Hochladen:\n%r") % (task,)) return new_id = client.wait_for_task(task) if choice.get("switch"): # Ab jetzt zeigt das Fenster auf das Zieldokument. Weitere # Versionen landen dort, nicht mehr beim Ursprung. cfg.set_document_id(doc, target_id, settings["base_url"]) url = doc.getURL() if url: cfg.remember(url, target_id, settings["base_url"]) hinweis = ("Das geöffnete Fenster ist jetzt dem Zieldokument " "zugeordnet." if choice.get("switch") else "Das geöffnete Fenster bleibt seiner bisherigen " "Zuordnung erhalten.") dlg.message(self.ctx, self.frame, "Als neue Version angehängt.\n\n" "Zieldokument: %s\nWurzel-ID: %d\nVersions-ID: %s\n" "Bezeichnung: %s\n\n%s" % ((target.get("title") or "")[:70], target_id, new_id, label, hinweis)) # -- Ziel: neues Dokument ------------------------------------------------ def _create_new(self, client, settings, doc, doc_id, source, choice, path, data): title = choice["title"] take_meta = choice.get("take_meta") fields = {"title": title[:127]} if take_meta and doc_id is not None: for key in ("correspondent", "document_type", "storage_path"): if source.get(key): fields[key] = source[key] if source.get("created"): fields["created"] = str(source["created"])[:10] task = client.post_document(os.path.basename(path), data, fields) if not isinstance(task, str) or not task: dlg.error(self.ctx, self.frame, t("Unerwartete Antwort beim Hochladen:\n%r") % (task,)) return new_id = client.wait_for_task(task) # Tags und Zusatzfelder gehen nicht über post_document, sondern # nachträglich per PATCH. Die Herkunftskennung wird dabei bewusst # NICHT kopiert - sie gehört zum Ursprungsdokument. if take_meta and doc_id is not None and new_id: payload = {} if source.get("tags"): payload["tags"] = source["tags"] if source.get("owner") is not None: payload["owner"] = source["owner"] cf = [f for f in (source.get("custom_fields") or []) if not self._ist_herkunftsfeld(client, settings, f)] if cf: payload["custom_fields"] = cf if payload: try: client.patch_document(new_id, payload) except PaperlessError as exc: dlg.message(self.ctx, self.frame, "Dokument angelegt, aber die Klassifizierung " "konnte nicht vollständig übernommen " "werden:\n\n%s" % exc) if choice.get("switch") and new_id: cfg.set_document_id(doc, new_id, settings["base_url"]) url = doc.getURL() if url: cfg.remember(url, new_id, settings["base_url"]) hinweis = ("Das geöffnete Fenster ist jetzt diesem Dokument " "zugeordnet." if choice.get("switch") else "Das geöffnete Fenster bleibt dem Ursprungsdokument " "zugeordnet.") dlg.message(self.ctx, self.frame, "Neues Dokument angelegt.\n\n" "Titel: %s\nPaperless-ID: %s\n\n%s" % (title[:70], new_id, hinweis)) # Ein frisch angelegtes Dokument trägt sonst nur den Titel. Ohne # Klassifizierung findet es später niemand wieder, und nachträglich # macht man es erfahrungsgemäß nicht mehr. Deshalb direkt anbieten. if new_id: self._classify(client, new_id) def _ist_herkunftsfeld(self, client, settings, field_entry): """ Zusatzfelder mit Herkunftsbezug nicht in Kopien übernehmen. Wer aus einem anderen System migriert hat, führt dort oft ein Feld mit der Fremdkennung. Eine Kopie mit derselben Kennung wäre falsch - es gäbe zwei Dokumente, die vorgeben, dasselbe Original zu sein. Welche Feldnamen das betrifft, steht in den Einstellungen unter "herkunftsfelder". """ muster = [str(m).lower() for m in (settings.get("herkunftsfelder") or [])] if not muster: return False try: if not hasattr(self, "_cf_names"): self._cf_names = {f["id"]: f["name"] for f in client.custom_fields()} name = (self._cf_names.get(field_entry.get("field")) or "").lower() except Exception: # noqa: BLE001 return False return any(m in name for m in muster) # ------------------------------------------- Befehl: Klassifizierung def cmd_classify(self): client, _settings = self._client() if client is None: return _doc, doc_id = self._current_doc_id() if doc_id is None: return self._classify(client, doc_id) def _classify(self, client, doc_id): """Klassifizierungsdialog für ein bestimmtes Dokument.""" try: remote = client.document(doc_id) tags = client.tags() corrs = client.correspondents() dtypes = client.document_types() except PaperlessError as exc: dlg.error(self.ctx, self.frame, t("Stammdaten konnten nicht geladen werden:\n%s") % exc) return payload = dlg.classify_dialog(self.ctx, self.frame, remote, tags, corrs, dtypes) if payload is None: return # Korrespondent und Dokumenttyp können als Freitext eingegeben # werden. Unbekannte Werte werden nach Rückfrage angelegt. payload["correspondent"] = self._resolve( client, "correspondents/", corrs, payload.pop("_correspondent_name", ""), "Korrespondent") payload["document_type"] = self._resolve( client, "document_types/", dtypes, payload.pop("_document_type_name", ""), "Dokumenttyp") client.patch_document(doc_id, payload) dlg.message(self.ctx, self.frame, "Klassifizierung gespeichert.") def _resolve(self, client, path, known, name, label): name = (name or "").strip() if not name: return None for row in known: if row["name"].strip().casefold() == name.casefold(): return row["id"] if not dlg.confirm(self.ctx, self.frame, "%s „%s“ existiert nicht.\n\nAnlegen?" % (label, name)): return None return client.create(path, name)["id"] # ---------------------------------------------------- Befehl: Teilen def cmd_share(self): """ Erzeugt einen zeitlich begrenzten, unauthentifizierten Link. Vorher wird geprüft, ob der geteilte Stand überhaupt im Archiv liegt. Ein Link auf eine Fassung, die nur lokal existiert, wäre irreführend - der Empfänger bekäme etwas anderes zu sehen, als der Absender vor sich hat. """ client, settings = self._client() if client is None: return doc, doc_id = self._current_doc_id(quiet=True) if doc is None: dlg.message(self.ctx, self.frame, "Kein Dokument geöffnet.") return # 1. Ist das Dokument überhaupt im Archiv? if doc_id is None: if dlg.confirm( self.ctx, self.frame, "Dieses Dokument liegt noch nicht in Paperless.\n\n" "Ein Freigabelink kann erst erzeugt werden, wenn es " "abgelegt ist.\n\nJetzt ablegen?"): self.cmd_save_as() return # 2. Gibt es lokale Änderungen, die noch nicht abgelegt sind? try: geaendert = bool(doc.isModified()) except Exception: # noqa: BLE001 geaendert = False if geaendert: antwort = dlg.choice( self.ctx, self.frame, "Nicht abgelegte Änderungen", "Das geöffnete Dokument enthält Änderungen, die noch nicht " "in Paperless abgelegt sind.\n\nDer Link würde auf die " "zuletzt abgelegte Fassung zeigen, nicht auf das, was du " "gerade vor dir hast.", ["Abbrechen", "Trotzdem teilen", "Erst neue Version ablegen"]) if antwort in (None, 0): return if antwort == 2: self.cmd_save_version() try: if doc.isModified(): return # Ablegen wurde abgebrochen except Exception: # noqa: BLE001 pass bekannte = [] try: bekannte = sorted(t["name"] for t in client.tags()) except PaperlessError: pass auswahl = dlg.share_dialog(self.ctx, self.frame, settings.get("share_days", 7), settings.get("share_tag", "Sharelink"), bekannte) if auswahl is None: return days, variant, tagname = auswahl url, roh = client.share_link(doc_id, days, variant) if not url: dlg.error(self.ctx, self.frame, "Der Link konnte nicht ermittelt werden.\n\n" "Antwort: %r" % (roh,)) return # 3. Kennzeichnung anfügen, ohne vorhandene Tags zu verlieren tag_hinweis = "" if tagname: try: tag_id = self._resolve(client, "tags/", client.tags(), tagname, "Tag") if tag_id: aktuell = client.document(doc_id).get("tags") or [] if tag_id not in aktuell: client.patch_document( doc_id, {"tags": list(aktuell) + [tag_id]}) tag_hinweis = "\nTag „%s“ gesetzt." % tagname # Für den nächsten Aufruf merken settings["share_tag"] = tagname cfg.save_settings(settings) except PaperlessError as exc: tag_hinweis = "\nTag konnte nicht gesetzt werden: %s" % exc gueltig = ("%d Tage" % days) if days else "unbegrenzt" dlg.result_dialog( self.ctx, self.frame, "Freigabelink erzeugt", "Der Link ist ohne Anmeldung abrufbar.\nGültigkeit: %s%s" % (gueltig, tag_hinweis), url) # ------------------------------------------ Befehl: Einstellungen def cmd_settings(self): current = cfg.load_settings() updated = dlg.settings_dialog(self.ctx, self.frame, current) if updated is None: return cfg.save_settings(updated) if not cfg.settings_complete(updated): return try: PaperlessClient(updated["base_url"], updated["token"], updated.get("verify_tls", True), user_agent=updated.get("user_agent") or None).ping() dlg.message(self.ctx, self.frame, "Verbindung erfolgreich hergestellt.") except PaperlessError as exc: dlg.error(self.ctx, self.frame, t("Gespeichert, aber die Verbindung schlug fehl:\n\n%s") % exc) # ------------------------------------------------------ Befehl: Info def cmd_info(self): client, settings = self._client(require=False) doc, doc_id = self._current_doc_id(quiet=True) lines = ["Basisadresse: %s" % (settings.get("base_url") or "—")] if doc is None: lines.append("Kein Dokument geöffnet.") elif doc_id is None: lines.append("Dieses Dokument stammt nicht aus Paperless.") elif client is None: lines.append("Paperless-ID: %d" % doc_id) else: try: r = client.document(doc_id) versions = r.get("versions") or [] lines += [ "Paperless-ID: %d" % doc_id, "Titel: %s" % (r.get("title") or "—"), "Archiv-Seriennummer: %s" % (r.get("archive_serial_number") or "—"), "Versionen: %d" % (len(versions) or 1), "Erstellt: %s" % ((r.get("created") or "")[:10] or "—"), ] except PaperlessError as exc: lines.append("Abruf fehlgeschlagen: %s" % exc) dlg.result_dialog(self.ctx, self.frame, "Paperless — Status", "\n".join(lines)) g_ImplementationHelper = unohelper.ImplementationHelper() g_ImplementationHelper.addImplementation(PaperlessAddon, IMPL_NAME, (SERVICE_NAME,))