#### Data Extraction
Um die gewünschten Analysen und Visualisierungen zu ermöglichen, wurden verschiedene Implementierungen geschaffen. Diese übernehmen einerseits den Bezug von Rohdaten aus verschiedenen Quellen aber auch die ersten Aufbereitungsschritte und letztlich Persistierung in der Staging DB.
##### Extraktion von News
Vor der Implementierung der News-Extraktion wurde zunächst analysiert, welche Informationen bzw. Attribute eines Artikels für die Applikation interessant sein könnten. Dabei wurden folgende Felder festgelegt:
- ID
Identifizierendes Merkmal für einen einzelnen Artikel, um Referenzen herstellen zu können bzw. Duplikate zu vermeiden – von der Quelle festzulegen.
- Title
Titel des Artikels im Klartext.
- Date
Veröffentlichungsdatum des Artikels im ISO 8601 Format.
- Text
Inhalt des Artikels im Klartext.
- Source URL
URL des Artikels, um Nutzer darauf verweisen zu können.
Nachdem das Zielformat der Newsartikel festgelegt wurde, bestand die Aufgabe, diese Informationen aus den ausgewählten Quellen zu extrahieren. Um eine einheitliche und leicht erweiterbare Lösung zu schaffen, wurde zunächst die folgende Software-Architektur geplant:
Über die Methode get_news_for_category wird ein Interface geschaffen, über welches die darüberliegende Applikation News aus der in den Kind-Klassen implementierten Quelle beziehen kann. So soll es möglich sein, in der Applikation keine Spezifika der Datenquellen implementieren zu müssen, sondern lediglich auf die bereits kapselnde Klasse zurückzugreifen.
Über den `BaseNewsExtractor` wird ferner eine geschützte Methode veranlasst, die es den Kind-Klassen ermöglicht, die von der jeweiligen Quelle bezogenen Daten zu parsen, aufzubereiten und letztlich in eine Liste von News-Objekten zu überführen. Mittels dieser Struktur könnten in Zukunft problemlos weitere Quellen angebunden werden, ohne große Änderungen am Verlauf der Haupt-Applikation vornehmen zu müssen. Es müsste lediglich die neue Klasse importiert und ein Objekt davon erstellt werden (Stichwort: "Programming To an Interface").
In der Applikation `fetch_news` wird darauf aufbauend die Überführung der von den `BaseNewsExtractor` erbenden Klassen bezogenen News in die Staging DB behandelt.
##### Stammdaten-Extraktion
Die Extraktion von Unternehmens-Stammdaten wurde zunächst gemäß des ETL-Lebenszykluses in drei Sub-Domänen unterteilt:
1. [Extract](https://github.com/fhswf/aki_prj23_transparenzregister/blob/main/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/extract.py)
Download der Rohdaten aus dem Unternehmensregister
2. [Transform](https://github.com/fhswf/aki_prj23_transparenzregister/tree/main/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/transform)
Aufbereitung und Bereinigung der Rohdaten sowie Überführung in das Zielformat
3. [Load](https://github.com/fhswf/aki_prj23_transparenzregister/blob/main/src/aki_prj23_transparenzregister/utils/data_extraction/unternehmensregister/load.py)
Einfügen bzw. Aktualisieren der Einträge in der Staging DB
Da das Unternehmensregister weder über eine API verfügt, noch bereits Python-Bibliotheken existieren, die den Umgang mit dem Unternehmensregister erleichtern, musste eine eigene Lösung von Grund auf entwickelt werden. Aufgrund der Beschaffenheit des Unternehmensregisters bestand die einzige Lösung in der Automatisierung eines Bots, der durch Klicks die Website des Unternehmensregisters steuert und sich so zum Download der gewünschten Informationen, dem sog. "Strukturierten Inhalt" – den Stammdaten eines Unternehmens in XML –, arbeitet.
Das Ziel dieses Schrittes war es, eben jene Dateien in einen dafür vorgesehenen Ordner auf dem ausführenden Client zu überführen, auf den dann im Folgeschritt zugegriffen werden kann.
Da die Laufzeit dieses Prozesses aufgrund der zwischengelagerten Warteintervalle auf der Website recht hoch war, wurde zunächst Abstand von einem kompletten Abbild des Unternehmensregisters genommen. Stattdessen wurde eine Verarbeitung der 100 größten deutschen Unternehmen laut Statista gestartet. Zusätzlich wurde die Verarbeitung so strukturiert und gekapselt, dass durch Multi-Processing mehrere Unternehmen parallel verarbeitet werden konnten. Hierzu wird für jede Suche eine eigene Instanz eines Headless Chrome Browsers und Selenium gestartet, so dass sich die Prozesse nicht gegenseitig in die Quere kommen können.
In der darauffolgenden Transformation wird jedes XML Dokument eingelesen und in das Zielformat der `Company` überführt:
Details können dem Quell-Code entnommen werden, im Groben werden Informationen wie folgt entnommen:
- ID
- HR Nummer: Aktenzeichen des Unternehmens
- Amtsgericht: 2. "Beteiligung" Eintrag
- Name: 1. "Beteiligung" Eintrag
- Location: 1. "Beteiligung" Eintrag
- Last Update: "Letzte Eintragung"
- Company Type: "Rechtsform" Eintag; alternativ: Suffix des Unternehmensnamens
- Capital: "Zusatzangaben" Eintrag
- Business Purpose: "Gegenstand oder Geschäftszweck" Eintrag
- Stakeholder: "Beteiligung" Eintrag >= 2
- Fouding Date: "Tag der ersten Eintragung" oder "Gesellschaftsvertrag vom" oder "Gruendungsdatum" Eintrag
Da viele der oben genannten Angaben nicht immer zur Verfügung stehen, wurden zahlreiche der Attribute optional bzw. als "nullable" gesetzt. Insgesamt wurden viele Probleme in den Daten identifiziert. Unter anderem sind Personen nicht immer eindeutig referenziert, sondern werden lediglich über ihren Namen und gegebenenfalls Geburtsdatum identifiziert. Dies stellt jedoch nur eine indirekte und möglicherweise unzuverlässige Methode zur Identifizierung dar.
Auch bei Adressen traten wiederholt Probleme auf, da Straßennamen in unterschiedlichen Einträgen unterschiedlich geschrieben wurden oder Hausnummer und Straße nicht sauber getrennt wurden. In der Implementierung sind daher zahlreiche vorgelagerte Überprüfungen und weitere Aufbereitungen zu finden, um die Daten sauber übergeben zu können.
Nach einiger Zeit jedoch traten plötzliche Fehler in den Routinen zur Extraktion von Stammdaten auf. Eine Analyse der Logs sowie der bezogenen und transformierten Daten zeigte, dass das Unternehmensregister nicht, wie ursprünglich angenommen, ein einheitliches Format pflegt. Stattdessen gibt es den sogenannten "Strukturierten Inhalt" zwar immer im XML-Format aus, jedoch kann das darunterliegende Schema je nach Alter des Eintrags unterschiedlich sein. Nähere Details zum Problem sind im Issue [Issue #233](https://github.com/fhswf/aki_prj23_transparenzregister/issues/233) zu finden.
xJustizVersion v1
```json
{
"XJustiz_Daten": {
"@xmlns": "http://www.xjustiz.de",
"@xmlns:xdo": "http://www.xdomea.de",
"@xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance",
"@xsi:schemaLocation": "http://www.xjustiz.de xj_0400_register_1_9.xsd",
"Grunddaten": {
"@XJustizVersion": "1.20.0",
...
}
```
xJustizVersion v3
```json
{
"tns:nachricht.reg.0400003": {
"@xmlns:tns": "http://www.xjustiz.de",
"@xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance",
"@xsi:schemaLocation": "http://www.xjustiz.de xjustiz_0400_register_3_2.xsd",
"tns:nachrichtenkopf": {
"@xjustizVersion": "3.3.1",
"tns:aktenzeichen.absender": null,
...
```
Nicht nur unterscheiden sich die Schlüssel zwischen den Versionen, sondern auch die Wertebereiche und der Einsatz von Enums zeigen signifikante Unterschiede.
Die Aufbereitungsroutine wurde daher überarbeitet. Ähnlich der Lösung mit dem `BaseNewsExtractor` aus der News-Verarbeitung wurde hier ebenfalls eine Oberklasse (`BaseTransformer`) eingeführt. Diese wird von den Implementierungen `v1` oder `v3` abgeleitet, die jeweils das gleichnamige Datenschema kapseln. Im BaseTransformer werden verschiedene Submethoden definiert, die jeweils einen Aufbereitungsschritt repräsentieren. Diese müssen von der entsprechenden Kindklasse gemäß des Datenschemas implementiert werden.
Die Methode `map_unternehmensregister_json` dient dabei als zentraler Einstiegspunkt für die darüberliegende Applikation. Generelle Funktionen wie zum Beispiel das Parsen von Datumsangaben oder die Normalisierung von Strings werden zentral zur Verfügung gestellt und können von allen Implementierungen verwendet werden.
Die `main`-Funktion des übergeordneten `transform`-Moduls bestimmt, welche Implementierung verwendet werden soll, indem sie den Header der Daten (Beispiele siehe oben) ausliest. Anschließend übergibt sie die Daten entweder an die `v1`- oder `v3`-Implementierung und speichert die Ergebnisse als JSON-Dateien in einem separaten Ordner.
Dieser Ordner wird schließlich von der Lade-Routine aufgegriffen, und jedes Dokument wird in die Staging DB überführt. Dabei werden neue Einträge – anhand der ID ermittelt – komplett neu angelegt, und Felder bestehender Artikel werden aktualisiert. Es findet kein vollständiger Austausch statt, der bereits ermittelte Finanzergebnisse überschreiben würde.
##### Extraktion von Finanzdaten
Finanzdaten und Kennzahlen der Unternehmen werden der Öffentlichkeit zentral über den [Bundesanzeiger](https://www.bundesanzeiger.de/) in Form von Jahresabschlüssen sowie der Veröffentlichung relevanter Bekanntmachungen, wie Änderungen im Aufsichtsrat oder der Terminierung des Unternehmens, zur Verfügung gestellt. Leider wird keine einfache API bereitgestellt, die es ermöglichen würde, gezielt gewünschte KPIs zu einem ausgewählten Unternehmen zu beziehen. Stattdessen existiert lediglich eine Python-Bibliothek namens [deutschland](https://github.com/bundesAPI/deutschland), in der eine Reihe von behördlichen Portalen gekapselt und bestimmte Interaktionen ermöglicht werden.
In dieser Bibliothek lässt sich bereits ein Paket finden, das für die Interaktion mit dem Bundesanzeiger zuständig ist. Dieses stellt wiederum eine Methode bereit, mit der eine Suche nach einem gegebenen Begriff durchgeführt und die Dokumente der **ersten** Seite ausgelesen werden können. Das resultierende Array enthält Datenobjekte mit Titel, Rohform des Inhalts (HTML) sowie dem reinen Text des Dokuments.
Die dabei gefundenen Elemente werden zunächst nach ihrem Typ gefiltert. Für den aktuellen Stand des Projekts sind lediglich die Jahresabschlüsse von Interesse. Nach dieser Filterung werden per Regex das Datum und das dazugehörige Geschäftsjahr extrahiert. Für jeden Eintrag werden dann die Wirtschaftsprüfer sowie die Finanzergebnisse extrahiert. Da die Angabe der Wirtschaftsprüfer über die bisher verarbeiteten Einträge konsistent ist, kann diese Information über die Struktur des Dokuments anhand der HTML-Struktur bezogen werden. Bei den Finanzdaten gestaltete sich dies etwas komplizierter:
Unter anderem ist das verwendete Format zwischen Unternehmen unterschiedlich, sowohl was die Strukturierung des Jahresabschlusses als auch die eingesetzten Fachwörter betrifft. So kann es sein, dass in einem Dokument der "Jahresumsatz in 1000€" im anderen als "Umsatz in EUR" ausgewiesen wird. In erster Instanz wurde eine Lösung entwickelt, die auf Basis von Regex in der Lage ist, eine Reihe von gewünschten Kennzahlen aus dem Text der Jahresabschlüsse zu extrahieren:
```python
kpi_patterns = {
FinancialKPIEnum.REVENUE.value: r"(?:revenue|umsatz|erlöse)[:\s]*([\d,.]+[mmb]?)",
FinancialKPIEnum.NET_INCOME.value: r"(?:net income|jahresüberschuss|nettoeinkommen|Ergebnis nach Steuern)[:\s]*([\d,.]+[mmb]?)",
FinancialKPIEnum.EBIT.value: r"(?:ebit|operating income)[:\s]*([\d,.]+[mmb]?)",
FinancialKPIEnum.EBITDA.value: r"(?:ebitda)[:\s]*([\d,.]+[mmb]?)",
FinancialKPIEnum.GROSS_PROFIT.value: r"(?:gross profit|bruttogewinn)[:\s]*([\d,.]+[mmb]?)",
...
}
```
Diese Variante übersetzt eine Reihe möglicher Begriffe in die im Projekt verankerten Kennzahlen-Enum der Modelle. Da viele Jahresabschlüsse die gewünschten Informationen jedoch nicht direkt im Text ausweisen, sondern diese in Tabellen visualisieren, konnte diese Lösung lediglich mäßige Erfolge erzielen.
Daher wurde eine weitere Lösung entwickelt, die nicht auf Basis des Textes, sondern auf Basis von in den Jahresabschlüssen aufgeführten Tabellen agiert. In dieser Implementierung werden jeglich im Jahresabschluss gefundenen Tabellen aufbereitet, transformiert und letztlich in ein Dictionary überführt, das alle Zeilenbezeichnungen, die numerische Werte führen, als Schlüssel und den dazugehörigen Wert – bereinigt auf eine einheitliche Einheit, den Euro – führt. Bei dieser Lösung handelt es sich im Vergleich zur vorherigen um einen etwas freieren Ansatz, der zwar in der Lage ist, deutlich mehr Kennzahlen zu extrahieren, diese aber noch nicht auf das spätere Ziel-Format zuschneidet.
Aufgrund der damit verbundenen Änderungen an den Folgekomponenten, wie dem `Data Processing`, wurden die bisher bezogenen Daten zunächst so belassen und noch nicht mit der neuen Lösung verarbeitet. In Zukunft könnte dies jedoch bessere Ergebnisse erzielen und vor allem durch eine Überführung in die Staging DB und die Trennung zur weiteren Verarbeitung eine Lösung schaffen, die bereits weitere Kennzahlen enthält und bei Bedarf in die weiteren Schichten führen kann..
##### Dockerization & Scheduling
Im Laufe der Entwicklung und mit der Integration des eigentlichen Deployments auf den Servern der FH SWF wurde deutlich, dass ein Scheduler für die Data Extraction Jobs über Apache Airflow den gesamten Deploymentsprozess und den damit verbundenen Organisationsaufwand erheblich erhöhen würde. Im Vergleich zur zuvor vorgestellten Airflow-Instanz, die auf einem Heimserver eines Projektmitglieds in einer Umgebung gehostet wurde, die keine weiteren Applikationen anderer Gruppen zur Verfügung stellt, wäre der Aufbau der Airflow-Instanz und vor allem die nachhaltige Konfiguration, einschließlich der Einrichtung eines entsprechenden User-Managements wie etwa die Anbindung an Keycloak, zu zeitaufwendig.
Da dennoch die Anforderung besteht, bestimmte Jobs der Datenextraktion regelmäßig zu starten, wurde ein dedizierter Docker Container mit einem integrierten Scheduler entwickelt.
Dieser Container, als `ingest` in den beiden `docker-compose`-Dateien des Projekts zu finden, greift auf bestehende Applikationen/Funktionen im Bereich der Data Extraction zu und kapselt diese im Scheduler. Der Scheduler stellt eine Erweiterung zum bereits existierenden [`schedule`](https://schedule.readthedocs.io/en/stable/index.html)-Paket dar und bietet zusätzlich die Funktion, die nächste Ausführung in einer Datei zu persistieren. Dadurch überlebt er beispielsweise das erneute Deployment der Applikation und startet mit dem gleichen Intervall ausgehend von der tatsächlich letzten Ausführung. Ohne diese Funktion würde das gesetzte Intervall nach jedem Deployment von Neuem starten. Dies hätte in der Hochphase der Entwicklung dazu geführt, dass das gewünschte Intervall nie erreicht werden kann, da in der Zwischenzeit immer wieder eine neue Version deployed wird.
Im Kern des `ingest` stehen nun zwei Ziele:
1. Das Beziehen aktueller News-Artikel aus den bereits vorgestellten Quellen (Tagesschau API, Handelsblatt RSS Feed)
2. Das Füllen von Lücken in der Staging DB in Bezug auf Unternehmen
Der erste Job wurde bereits ausführlich in einem vorherigen Kapitel erläutert. Beim zweiten Job handelt es sich um eine Routine aus verschiedenen Schritten. Neben dem Bezug von Finanzdaten für Unternehmen, für welche noch keine solchen Daten in der Staging DB vorliegen, liegt das Hauptaugenmerk auf der Stammdaten-Extraktion für Unternehmen. Diese Unternehmen werden während der Daten-Transformation von der Staging- in die ProdDB zwar referenziert (zum Beispiel über Wirtschaftsprüfer-Beziehungen oder als Prokuristen etc.), sind jedoch als eigenständige Einträge noch nicht vorhanden. Dieser Job greift auf die bestehenden Scraping-Funktionen zurück und nutzt diese, um eben jene Daten zu beziehen. Durch den erneuten Aufruf der Daten-Transformationsroutine kann die Datenbasis so natürlich wachsen und erweitert sich täglich automatisch, ohne dass manuelles Eingreifen nötig wäre.
Auch wenn die Eigenimplementierung des Schedulers im Vergleich zu den Möglichkeiten von Apache Airflow einen bedeutenden Rückschritt darstellt, da unter anderem grafisch aufbereitete Logs und Fehleranalysen verloren gehen, konnte durch diese Entscheidung im Rahmen des Projektes – welches lediglich den Charakter eines Proof of Concepts (PoC) genießt – Zeit sowie Aufwand eingespart werden. Auch die daher gesparten Entwicklungsaufwände, um den bestehenden Code in das von Airflow gewünschte Format zu überführen, entfielen.
##### Zwischenfazit
Zusammenfassend lässt sich festhalten, dass die anfänglich gesetzten Ziele der Datenextraktion nur teilweise erfüllt werden konnten. Zwar sind die grundlegenden Informationen wie Stammdaten sowie die ersten Finanzergebnisse vorhanden, jedoch fehlen viele der initial festgelegten Kennzahlen, die sich auch mit dem aktuellen Datenbestand nur partiell in der "Data Processing"-Komponente berechnen lassen. Ebenfalls fehlt grundlegend eine Historisierung der Daten, sodass immer nur eine Momentaufnahme erstellt werden kann. Sollte das Projekt weiter fortgeführt werden, so muss hier angesetzt werden, um auch den Verlauf der am Unternehmen beteiligten Personen oder weitere relevante Änderungen nachverfolgen zu können.
Nichtsdestotrotz wurde im Rahmen der Entwicklung Code geschaffen, der sich dafür eignet, in das "deutschland"-Paket als Wrapper für das Unternehmensregister eingeführt zu werden. Dass der Bedarf für eine solche Funktion besteht, zeigt ein aktuelles Issue [#132](https://github.com/bundesAPI/deutschland/issues/132).
Ebenfalls anzumerken ist die technische Hürde, die aktuell von Behörden gestellt wird, um an die gewünschten Daten zu gelangen. Es wäre wünschenswert, in Zukunft eine einfacher zugängliche Lösung – z.B. auf Basis öffentlicher APIs – zu sehen.