Import-Services

Einleitung

docs365 documents unterstützt verschiedene Hotfolder-Typen, um Daten aus z.B. PDF-Dateien, PHOENIX-Capture-Exporten oder bestimmten XML-Dateien ins Archiv zu importieren. Diese Hotfolder-Typen können durch Konfiguration an verschiedene Anwendungsfälle angepasst werden. Datenimporte aus eigenen, speziellen Formaten übersteigen jedoch die Möglichkeiten der mitgelieferten Hotfolder-Typen, sodass individuelle Import-Skripte notwendig sind. Diese Funktionalität wird durch die Import-Services der Hotfolder bereitgestellt.

Ein Import-Service ist ein spezieller Hotfolder-Typ, der die Verarbeitung der Dateien des Hotfolders an ein eigenes, in Python 3.7 programmiertes Plugin weiterleitet. Konfiguriert wird ein Import-Service in der hotfolder.ini folgendermaßen:

[Demo-Archiv]
Directory=D:\Hotfolder
Type=importservice
Filemask=*.xml
Module=myimport

Type=importservice definiert, dass es sich bei diesem Hotfolder um einen Import-Service handelt. Es werden in diesem Beispiel alle *.xml-Dateien im Verzeichnis D:\Hotfolder verarbeitet und an das Python-Modul myimport weitergeleitet.

Die Python-Module müssen im Unterverzeichnis customisation\importservices des Installations-Verzeichnisses abgelegt werden und die Endung .py haben. Wenn docs365 documents z.B. in C:\Programme\PHOENIX_Documents installiert ist, befindet sich der Import-Service myimport in der Datei C:\Programme\PHOENIX_Documents\customisation\importservices\myimport.py.

Jeder Import-Service kann durch eine Konfigurationsdatei weiter angepasst werden. Wenn sich neben der Datei myimport.py noch eine INI-Datei myimport.ini befindet, wird diese automatisch eingelesen und im Python-Modul bereitgestellt.

Aufbau eines Import-Services

Das folgende Beispiel ist ein minimaler Import-Service. Das Python-Modul muss eine Klasse mit dem Namen ImportService enthalten, die von der Basisklasse importservice.Base ableitet.

import importservice

class ImportService(importservice.Base):
    """
    ImportService-Vorlage
    """
    def process(self, file, result):
        pass

    # optional
    def handle_generic_error(self, error):
        pass

    # optional
    def handle_client_error(self, error):
        pass

    # optional
    def post_process(self, record):
        pass

    # optional
    def pre_process(self):
        pass

Die Methode process wird für jede im Hotfolder zu verarbeitende Datei einzeln aufgerufen. Der Parameter file ist der vollständige Pfad der aktuellen Datei und result ist ein Objekt, über das man mithilfe von verschiedenen Methoden den Archivierungvorgang steuern kann:

result.archive

In dieses Attribut wird die ID oder der Identifier des Archivs geschrieben, in das archiviert werden soll. Analog zum Archive-Parameter des Hotfolders, ist auch hier folgendes möglich:

result.archive = '@schmidt'
result.archive = 'invoice@schmidt'
result.archive = '#Gruppenname'
result.archive = 'invoice#Gruppenname'

Wird dieses Attribut nicht gesetzt, wird stattdessen die Direktive Archive=... aus der hotfolder.ini verwendet. Fehlt diese ebenfalls, wird der Archivierungvorgang mit einer Fehlermeldung abgebrochen.

result.metadata

Ein dict, in das die Feldwerte des Vorgangs geschrieben werden. Der Schlüssel ist dabei der Identifier der jeweiligen Spalte. In dem Demo-Import-Service weiter unten gibt es Beispiele für mögliche Werte verschiedener Typen.

result.skip

Wenn dieses Attribut auf True gesetzt wird, wird die aktuell verarbeitete Datei übersprungen und im Hotfolder liegen gelassen, sodass sie bei einem zukünfigen Aufruf noch einmal verarbeitet wird.

result.postbox

Dieses optionale Attribut ist eine Liste, in der Benutzer als Strings hinzugefügt werden können. Sie gibt an welche Benutzer einen Postfach-Eintrag des importierten Vorganges erhalten. Wird dieses Attribut nicht gesetzt, wird stattdessen die Direktive Postbox=... aus der hotfolder.ini verwendet.

result.client

Ein optionales Attribut, das zur Zuweisung der Mandanten-Fähigkeit benötigt wird. Der Attribut-Typ ist ein String, welcher der Kürzel des zugeordneten Madanten eines Archives ist. Wird dieses Attribut nicht gesetzt, wird stattdessen die Direktive Client=... aus der hotfolder.ini verwendet.

result.add_file(file, *, filename=None, custom_name=None, custom_description=None, custom_date=None, insert=None, replace=None)

Über diese Methode werden Dokumentenanhänge an den Vorgang angehängt. file ist der Dateiname relativ zum Hotfolder, es kann aber auch ein absoluter Pfad angegeben werden. Mit dem optionalen Parameter filename kann der Dateiname im Archiv angepasst werden. Fehlt dieser Parameter, wird der Dateiname von file benutzt. Wenn die Hauptdatei (der file-Parameter der process-Methode) selbst auch archiviert werden soll, muss sie hier ebenfalls angegeben werden.

Die Verwendung der insert und replace Parameter werden im Abschnitt Hinzufügen und Ersetzen von Dokumenten im Vorgang näher erläutert.

Eine bereits existierende Historie sowie Bemerkungen können im Import-Service ebenfalls übergeben werden:

result.add_history(action, timestamp=None, username='[import]'): Fügt einen Historien-Eintrag hinzu.
result.add_remark(text, timestamp=None, username='[import]'): Fügt eine Bemerkung hinzu.

Die beiden Methoden add_history und add_remark können mehrfach aufgerufen werden, um mehrere Einträge zu übergeben. Der jeweils erste Parameter (action bzw. text) muss vom Typ str sein. timestamp versteht ein Datum (mit oder ohne Uhrzeit) im ISO-8601-Format oder einen nativen date- der datetime-Typ von Python. Wenn nur ein Datum angegeben ist, wird als Uhrzeit Mitternacht der lokalen Zeitzone angenommen. Wenn timestamp gleich None ist, wird als Vorgabewert die aktuelle Uhrzeit eingetragen. username ist ebenfalls optional und muss ein str sein. Wenn kein username übergeben wird, wird als Vorgabe [import] eingetragen. Wenn einer der Parameter einen ungültigen Wert hat, wird als Exception ein ValueError ausgelöst.

Bei einer erfolgreichen Hotfolder-Verarbeitung kann die optionale Methode post_process dazu verwendet werden, nach dem Archivieren auf den archivierten Vorgang zuzugreifen, um beispielsweise die Vorgangs-ID in einer externen Datenbank abzuspeichern. Es wird die gleiche Objekt-Instanz wie in der process-Methode verwendet, sodass es möglich ist, sich in der self-Variable Informationen zwischen diesen beiden Methodenaufrufen zwischenzuspeichern. Wenn in der post_process eine Exception ausgelöst wird, wird der Archivierungs-Vorgang nicht abgebrochen, da er zu diesem Zeitpunkt schon abgeschlossen wird.

Die Methode pre_process ist optional und wird einmalig aufgerufen, bevor die erste Datei verarbeitet wird.

Das folgende Beispiel erzeugt einen Vorgang im Archiv demoarchiv mit zwei Feldwerten und einem Dokumentenanhang:

import importservice
import os
import xml.etree.ElementTree as etree

class ImportService(importservice.Base):
    """
    Demonstration aller Funktionen
    """
    def process(self, file, result):
        # Es soll in ein Archiv mit der ID 'demoarchiv' archiviert
        # werden.
        result.archive = 'demoarchiv'

        # Das Archiv 'demoarchiv' ist einem Mandanten 'demomandant'
        # zugewiesen
        result.client = 'demomandant'

        # Die Datei aus dem `file`-Parameter soll selbst archiviert
        # werden und muss daher explizit hinzugefügt werden.
        result.add_file(file)

        # `file` wird als absoluter, vollständiger Dateipfad übergeben.
        # Sollten Feldinformationen im Dateinamen vorhanden sein, kann
        # der Dateiname über os.path.basename() extrahiert werden.
        filename = os.path.basename(file)

        # Wenn `file` eine XML-Datei ist, kann sie mit dem
        # ElementTree-Module geparst werden:
        # https://docs.python.org/3.4/library/xml.etree.elementtree.html
        xml = etree.parse(file)

        # Schreibe Werte in die Archiv-Spalten
        result.metadata['text'] = 'Lorem ipsum'
        result.metadata['zahl'] = 12345
        result.metadata['bool'] = True

        # Datumswerte werden im ISO-8601-Format übergeben.
        result.metadata['date'] = '2015-02-09'
        result.metadata['datetime'] = '2015-02-09T02:54:51+01:00'

        # Fügt zusätzlich die Datei test.pdf aus dem Hotfolder hinzu.
        result.add_file('test.pdf')

        # Fügt einen Historieneintrag hinzu.
        result.add_history('Geprüft', '2017-12-18T14:43:28+00:00', 'Petra Prüfer')

        # Fügt eine Bemerkung hinzu.
        result.add_remark('Hallo', '2017-12-18', 'Thorsten Techniker')

    def post_process(self, record):
        # Gibt die ID und den Dateinamen aller angehängten Dokumente aus.
        for attachment in record['attachments']:
            print(attachment['id'], attachment['filename'])

    # Fehlertext (String) und Statuscode (Integer)
    def handle_generic_error(self, error):
        print(error.message, error.status)

    def handle_client_error(self, error):
        print(error.message, error.status)

Nach einem erfolgreichen Archivierungsvorgang werden sowohl die Hauptdatei file als auch alle über result.add_file hinzugefügten Dateien in das Unterverzeichnis _Completed verschoben.

Abbruch eines Archivierungsvorgangs

Sobald die process-Methode abgeschlossen ist und dem Vorgang ein Archiv eindeutig zugewiesen werden kann, wird die eigentliche Archivierung durchgeführt. Tritt innerhalb der process-Methode eine beliebige Exception auf, wird der Archivierungsvorgang abgebrochen und sowohl die Hauptdatei file als auch alle über result.add_file hinzugefügten Dateien in das Unterverzeichnis _Failed verschoben.

Wenn der Archivierungsvorgang manuell abgebrochen werden soll, sollte mit raise RuntimeError('Fehlermeldung') eine Exception ausgelöst werden. Die in der Exception enthaltene Fehlermeldung wird dabei ins Systemlog geschrieben.

Fehlerbehandlung

Sollte während der Archivierung ein Fehler auftreten, wird im Import-Service eine von zwei Methoden handle_generic_error und handle_client_error zur Fehlerbehandlung aufgerufen.

Generische Fehler

def handle_generic_error(self, error):
    print(error.message, error.status)

Ein Fehler, der auftritt falls während des ImportServices ein Timeout oder Authentifizierungsfehler entsteht. Serverfehler (Status-Codes 500-599) zählen ebenfalls zu dieser Kategorie. Die Fehlernachricht wird anhand error.message aufgerufen und der Statuscode, falls vergeben, mit error.status.

Generische Fehler sind üblicherweise nur vorübergehend (z.B. bei einem Timeout oder einem Serverfehler), daher werden die vom Import-Service verarbeiteten Dateien in diesem Fall zurück in den Hotfolder verschoben, sodass die Verarbeitung später erneut durchgeführt wird. Eine Ausnahme bilden Fehler mit dem Statuscode 500. In diesem Fall werden die Dateien in das _Failed-Verzeichnis verschoben.

Clientseitige Fehler

def handle_client_error(self, error):
    print(error.message, error.status)

Diese Methode wird aufgerufen, wenn die Archivierung aufgrund von clientseitigen Fehlern nicht durchgeführt werden konnte. Dazu zählen beispielsweise ein fehlerhaftes Mandanten-Kürzel oder ein nicht ausgefülltes Pflichtfeld.

error.message enthält den Fehlertext.
error.status enthält den vom Server zurückgegebenen Status-Code oder None, wenn die Archivierung aufgrund von fehlerhaften Parametern bereits client-seitig abgebrochen wurde.

Bei clientseitigen Fehlern werden die vom Import-Service verarbeiteten Dateien in das _Failed-Verzeichnis verschoben und die Fehlermeldung in das Ereignislog des Servers geschrieben.

Konfiguration

Die zu myimport.py gehörende Konfigurationsdatei myimport.ini kann in der Klasse ImportService über self.config benutzt werden. self.config ist ein configparser.ConfigParser-Objekt (aus der Python-Standard-Library) und ist leer, falls keine INI-Datei vorhanden ist.

Über self.root_config kann auf die Konfiguration des aktuellen Import-Services in der hotfolder.ini zugegriffen werden. self.root_config ist ein Objekt mit den folgenden Eigenschaften:

name: Der Name der Sektion in der hotfolder.ini
directory: Das absolute Verzeichnis des aktuellen Hotfolders.
items: Ein Python-dict, das unverändert alle Werte aus der Sektion in der hotfolder.ini enthält. Alle Schlüssel in diesem dict sind komplett klein geschrieben.
subsections: Ein Python-dict, das alle Untersektionen aus der hotfolder.ini enthält. Wenn es z.B. zu dem Hotfolder [Import] eine Untersektion [Import/Values] gibt, enthält dieses dict einen Schlüssel values (wieder komplett klein geschrieben), unter dem es ein weiteres dict mit den Wertepaaren aus dieser Untersektion gibt.

Generischer HTTP-Client

Der Import-Service stellt über self.client einen generischen HTTP-Client bereit. Dieser greift per Dienstbenutzer über API-Endpunkte auf den Server zu und erfordert daher eine konfigurierte Rolle. Weitere Anwendungsmöglichkeiten werden weiter unten unten aufgeführt.

Datenbankzugang

Es ist möglich, innerhalb der process-Methode über ODBC auf externe Datenbanken zuzugreifen. Eine ODBC-Datenquelle kann über die Sektion Database in der zum Modul gehörenden INI-Datei einfach konfiguriert werden:

[Database]
DSN = MY_DSN
UID = user
PWD = password

Anschließend kann über self.database mithilfe des with-Statements eine Verbindung aufgebaut werden:

with self.database() as conn:
    sql = 'SELECT data FROM mydata WHERE ID = {}'.format(documentId)
    cursor = conn.cursor()
    cursor.execute(sql)
    value = cursor.fetchone()[0]
    # ...

Sollten Daten verändert werden, muss die Transaktion mit conn.commit() abgeschlossen werden. Beim Verlassen des with-Blocks wird die Datenbankverbindung automatisch geschlossen.

Intern wird für die Datenbankverbindungen PyODBC benutzt. Sollte innerhalb eines Archivierungsvorgangs Zugriff auf mehr als eine externe Datenbank benötigt werden, muss PyODBC direkt verwendet werden.

Wichtig: Da docs365 documents eine 64-Bit-Anwendung ist, werden aktuell 64-Bit-ODBC-Treiber vorausgesetzt.

Zusätzliche Funktionen

self.run_smartindexing(archive, filename, client=None, ocr=False, on_smartindexing_event=False)

Führt SmartIndexing für die Datei filename (relativ zum Hotfolder-Verzeichnis) und das Archiv mit dem Kurznamen archive aus. Der Rückgabewert ist ein dict mit dem SmartIndexing-Ergebnis, welches auf Basis des übergebenen Mandanten client ermittelt wurde.

Optional kann das on_smartindexing-Event ausgeführt werden. Die vom Event ermittelten Werte sind bereits im SmartIndexing-Ergebnis enthalten.

Wird der Mandant nicht als Parameter angegeben, wird stattdessen die Direktive Client=... aus der hotfolder.ini verwendet.

Optional kann OCR auf die Datei angewandt werden. Mit dem Parameter ocr=True wird dies erzwungen. Standardmäßig ist dieser auf False gesetzt.

self.run_smartclassify(filename)

Führt SmartClassify für die Datei filename (relativ zum Hotfolder-Verzeichnis) aus. Wenn SmartClassify ein Archiv zuordnen konnte, wird dessen Kurzname zurückgegeben, ansonsten None.

self.run_zux_import(file, config, filename=None)

Führt einen ZUX-Import (ZUGFeRD- und XRechnungen) für eine PDF- oder XML-Datei file aus. Der Rückgabewert ist ein dict mit den ermittelten Rechnungsdaten, basierend auf der übergebenen Konfiguration config. Ein alternativer Dateiname filename kann angegeben werden, falls beispielsweise der Name der Import-Datei vom Standard abweicht - z.B. falls es sich um eine temporäre Datei ohne Dateiendung handelt.

Eine Konfiguration kann wie folgt aufgebaut sein. In diesem Format werden die ermittelten Daten von der Methode zurückgegeben.

{
  "ZUX_Fields": {
    "_kontierung": "<KONTIERUNG_POS>",
  },
  "ZUX/Fields/KONTIERUNG_POS": {
    "_betrag": "F,<BETRAG>",
    "_kostenstelle": "<KOSTENSTELLE>",
    "_steuersatz": "F,<STEUERPROZ>",
    "_kostenstelle": "<KOSTENSTELLE>",
    "_buchtext": "<BUCHTEXT_2>[255]",
  },
}

Existierende Vorgangswerte einer Spalte laden

Im Import-Service wird die Möglichkeit angeboten sich zu einem Datenfeld eines Archivs eine Liste zugehöriger Vorgangs-IDs zu laden. Beim Aufruf der Methode wird ein Archiv und ein Datenfeld-Kürzel angegeben, anhand dessen alle Vorgangs-IDs in einer Liste gruppiert zu einem einzigartigen Wert zurückgegeben werden. Eine dafür geeignete Stelle ist zum Beispiel im pre_process.

class ImportService(importservice.Base):
    def pre_process(self):
        self.existing = self.get_records_fieldmapping("invoices", "invoice_nr")

    def mapped_fields(self, file):
         mapping = {}
         # Logik zum Befüllen des Mappings anhand der Import-Datei

        return mapping

    def process(self, file, result):
         # Gemappte Daten aus einer eigenen Funktion laden
         self.data = self.mapped_fields(file)

         if self.data["I6011000-1"] in self.existing:
             # Logik nach dem die Prüfung erfolgreich war
             pass

Die Methode get_records_fieldmapping gibt ein Dictionary im folgenden Format zurück:

{
   "Beleg-Nr. 1": ["FFFFFFFF-0000-0000-0000-000000000002"],
   "Beleg-Nr. 2": ["FFFFFFFF-0000-0000-0000-000000000001", "FFFFFFFF-0000-0000-0000-000000000003"]
}

Existierende Vorgänge über Import-Services modizifieren

Import-Services können auch dazu verwendet werden, über einen Hotfolder Dokumente an bereits bestehende Vorgänge anzuhängen. Dafür muss im result-Objekt anstatt result.archive die ID des zu erweiternden Vorgangs in result.record geschrieben werden.

Über result.metadata können auch die Spaltenwerte des Vorgangs modifiziert werden (sofern die Spalten als Änderbar konfiguriert sind), das Format ist hier das Gleiche wie beim Anlegen von neuen Vorgängen.

Der ImportService stellt eine Hilfsfunktionen self.find_records(archive, query) bereit, mit denen nach Vorgängen gesucht werden kann. Der Parameter archive ist die ID oder der Kurzname des zu verwendenen Archivs und query ein Such-Filter mit der gleichen Syntax, die auch in den gefilterten Archiven oder der Spaltensuche verwendet wird. Der Rückgabewert ist eine Liste von gefundenen Vorgängen als dict, die ID kann über den Schlüssel id extrahiert werden:

class ImportService(importservice.Base):
    def process(self, file, result):
        result.add_file(file)

        # Vorgang im Archiv "rechnungen" suchen, dessen
        # Belegnummer "ABC123" ist.
        records = self.find_records('rechnungen', '_belegnr = "ABC123"')
        if len(records) != 1:
            raise RuntimeError('Es wurde genau ein Vorgang erwartet.')
        result.record = records[0]['id']

Beim Aufruf von find_records sollte grundsätzlich die Länge der zurückgegebenen Liste geprüft werden, da bei Filtern auch gar keine oder mehrere Vorgänge zurückgegeben werden. Die maximale Anzahl der gefundenen Vorgänge ist auf 100 beschränkt.

Hinzufügen und Ersetzen von Dokumenten im Vorgang

Standardmäßig wird beim Aufruf der Funktion result.add_file() die übergebene Datei ans Ende der Dokumente-Liste des modizifierten Vorgangs angehängt.

Der Keyword-Parameter insert akzeptiert einen int-Wert und kann dafür verwendet werden, die Position des neuen Dokuments in der Dokumenten-Liste festzulegen. Das erste Dokument beginnt dabei an Position 0, wenn man diesen Wert übergibt, wird das neue Dokument also an erster Stelle hingefügt (und alle weiteren Dokumente in der Liste um eine Position verschoben):

# An erster Position bestehender Anhänge hinzufügen
result.add_file(file, insert=0)

Der Parameter replace ersetzt ein bestehendes Dokument im Vorgang. Wenn man einen int-Wert übergibt, wird (analog zum insert-Parameter) das Dokument an der angegeben Stelle ersetzt.

Alternativ kann man bei replace auch einen str-Wert übergeben, der den Dateinamen des zu ersetzenden Dokuments enthält:

# Dokument an zweiter Position ersetzen
result.add_file(file, replace=1)

# Anhang "rechnung_20220920.pdf" durch Import-Datei ersetzen
result.add_file(file, replace="rechnung_20220920.pdf")

OCR-Text aus Datei

Im Import-Service wird die Methode get_file_text() zur Verfügung gestellt. Diese Methode extrahiert Text aus einer Datei, sei es ein Bild oder Textdokument. Wenn der Parameter ocr=True verwendet wird, erfolgt zusätzlich eine optische Zeichenerkennung (OCR). Der Rückgabewert kann ein Text oder None sein.

class ImportService(importservice.Base):
    def process(self, file, result):
        text = self.get_file_text(file, ocr=True)

Datei-Export

Um ein Dokument zu exportieren, muss die Methode self.write_attachment() verwendet werden, die mit folgenden Parametern aufgerufen werden kann:

self.write_attachment(attachment, path, on_duplicate="enumerate")

attachment: Das Dokument, das exportiert werden soll, übergeben als dict aus record['attachments'][n], wobei n in Listen-Index ist.
path: Wahlweise ein Verzeichnis oder ein vollständiger Dateipfad. Dieser Pfad muss absolut sein. Wenn als Pfad ein Verzeichnis angegeben wird, wird das Dokument unter dem im Archiv hinterlegten Dateinamen in diesem Verzeichnis abgespeichert.
on_duplicate: Gibt an, was passieren soll, wenn die in path übergebene Datei bereits existiert. Mögliche Werte sind enumerate (es wird ein Zähler an die Datei angehängt, z.B. datei_01.pdf), overwrite (die Datei wird überschrieben) und skip (die Datei wird übersprungen und nicht exportiert). Vorgabe ist enumerate.

Die Funktion liefert den absoluten Pfad zurück, unter dem die Datei abgespeichert wurde.

Beispiel:

class ImportService(importservice.Base):
 def process(self, file, result):
     result.add_file(file)

 def post_process(self, record):
     record = self.client.jget("records/" + record["id"])
     self.write_attachment(record["attachments"][0], r"D:\Documents"))

In diesem Beispiel wird die verarbeitete PDF automatisch zum Vorgang hinzugefügt und anschließend wieder in ein lokales Verzeichnis heruntergeladen.

Workflow-Übergänge ausführen

Es ist möglich, beim Importieren eines Vorgangs automatisch den nächsten Status-Übergang durchzuführen. Der Übergang darf dabei keine Benutzer-Rückfragen machen. Der Import-Service bietet dafür eine Methode mit folgender Signatur an.

def execute_transition(self, record, *, id=None, name=None)

Der record-Parameter aus der post_process-Methode wird zum Laden des neuen Vorgangs benötigt. Anhand des serialisierten Dictionaries (geladen durch den HTTP-Client) wird der Methode execute_transition() ein Kontext bereitgestellt. Die üblichen Parameter sind optional, solang es exakt einen Übergang gibt. Bei mehreren möglichen Übergängen ist es sinnvoll, den nächsten Übergang anhand des Names im Parameter name anzugeben.

Beispiel:

class ImportService(importservice.Base):
     def post_process(self, record):
         record = self.client.jget("records/" + record["id"])
         self.execute_transition(record, name="Workflow-Übergangsname")