read_transactions.webcrawler.base¶

author:: Tim Häberlein
version:: 2.2
date:: 21.10.2025
organisation:: TU Dresden, FZM

WebCrawler¶

Zentrale Basisklasse für alle Crawler im Projekt read_transactions_fm. - Einheitliches Logging über MainLogger - Nutzung einer externen WebDriverFactory - Robuste Typprüfung & flexible Datumsübergabe - Standardmethoden für Login, Download, Verarbeitung und Speicherung

Classes¶

WebCrawler

Abstrakte Basisklasse für alle Crawler im Paket read_transactions.

Module Contents¶

Abstrakte Basisklasse für alle Crawler im Paket read_transactions.

Diese Klasse kapselt den gesamten gemeinsamen Funktionsumfang: - zentrale Logging-Initialisierung über MainLogger - automatisches Laden von Konfiguration und Zugangsdaten aus config.yaml - standardisierte Selenium-WebDriver-Erzeugung über WebDriverFactory - konsistente Handhabung von Downloads, Datenverarbeitung und Speicherung

Subklassen (z. B. ArivaCrawler, AmazonCrawler) müssen mindestens die Methoden login() und download_data() überschreiben.

Typischer Ablauf:

```python with MyCrawler(start_date=“01.01.2024“, end_date=“31.03.2024“) as crawler:

crawler.login() crawler.download_data() crawler.process_data() crawler.save_data()

```

Parameter¶

namestr, optional: Logisch eindeutiger Name der Crawler-Instanz (z. B. "ariva").
output_pathstr, optional: Verzeichnis, in dem Ausgabedateien gespeichert werden (Standard: out).
start_datestr | pandas.Timestamp | datetime.date, optional: Startdatum für den Datenabruf.
end_datestr | pandas.Timestamp | datetime.date, optional: Enddatum für den Datenabruf.
detailsbool, optional: Ob zusätzliche Details extrahiert werden sollen? Uum beispiel bei trade_republic zusätzliche Order-Details oder bei amazon_visa vgl. mit amazon käufen. (Standard: True).
logging_levelstr, optional: Log-Level der Instanz (z. B. „DEBUG“, „INFO“, „WARNING“). Standard: INFO.
global_log_levelstr, optional: Globales Log-Level für das gesamte Paket (Standard: INFO).
logfilestr, optional: Pfad zu einer zentralen Logdatei (Standard: logs/read_transactions.log).
browserstr, optional: Verwendeter Browser-Treiber (edge, chrome oder firefox). Standard: edge.
headlessbool, optional: Aktiviert Headless-Modus (sofern vom Browser unterstützt). Standard: False.
user_agentstr, optional: Optionaler benutzerdefinierter User-Agent.

Attribute¶

driverselenium.webdriver.Remote: Aktiver Selenium-WebDriver.
datapandas.DataFrame | dict[str, pandas.DataFrame]: Heruntergeladene bzw. verarbeitete Daten.
_credentialsdict: Login-Daten des Crawlers (aus config.yaml).
_urlsdict: URL-Mappings des Crawlers (aus config.yaml).
_loggerlogging.Logger: Instanzspezifischer Logger.
_download_directorystr: Temporäres Verzeichnis für heruntergeladene Dateien.

__name = 'WebCrawler'¶

__output_path = 'out'¶

_logging_lvl = 'INFO'¶

__logger¶

property start_date: pandas.Timestamp¶: Startdatum (immer als pandas.Timestamp gespeichert).

property end_date: pandas.Timestamp¶: Enddatum (immer als pandas.Timestamp gespeichert).

property with_details: bool¶

Gibt zurück, ob zusätzliche Details extrahiert werden:

Bei Trade Republic z. B. zusätzliche Order-Details
Bei Amazon Visa z. B. Verknüpfung mit Amazon-Käufen

_state = 'initialized'¶

_download_directory = b'.'¶

_initial_file_count = 0¶

__credentials: Dict[str, str]¶

__urls: Dict[str, str]¶

__data: pandas.DataFrame | Dict[str, pandas.DataFrame]¶

__account_balance = 0.0¶

driver¶

property name: str¶: Name der Crawler-Instanz.

property data: pandas.DataFrame | Dict[str, pandas.DataFrame]¶: Heruntergeladene und ggf. verarbeitete Daten.

property _credentials: Dict[str, str]¶: Login-Daten für den Crawler. Beinhaltet z. B. Benutzername und Passwort. - user: str, Benutzername - password: str, Passwort

property _urls: Dict[str, str]¶: URLs für den Crawler. Beinhaltet z. B. Login- und Download-Links. - login: str, Login-URL - transactions: str, Transaktions-URL - kurse: Dict[str, str]: Kurs-URLs

property _logger: logging.Logger¶: Interner Logger (für Subklassen).

property account_balance: str¶: Gibt den aktuellen Kontostand zurück.

login() → None¶: Wird von Subklassen überschrieben – führt Login auf der Webseite aus.

download_data() → None¶: Wird von Subklassen überschrieben – startet Download-Vorgang.

process_data(read_temp_files: bool = True, sep: str = ';') → None¶

Optional von Subklassen überschreiben – verarbeitet geladene Daten. Standardmäßig werden alle Dateien im temporären Download-Verzeichnis eingelesen und in ein einziges DataFrame zusammengeführt. Dabei wird die Funktion preprocess_data() für jedes DataFrame aufgerufen.

Im Anschluss wird self.data normalisiert.

Parameter:

read_temp_files (bool, optional) – Ob Dateien im temporären Download-Verzeichnis eingelesen werden sollen.
sep (str, optional) – Trennzeichen für CSV-Dateien. Standard ist ‚;‘.

Rückgabe:

None

preprocess_data(key: str, df: pandas.DataFrame) → pandas.DataFrame¶

Bereinigt ein einzelnes DataFrame. Für weitere Verarbeitung muss funktion in Unterklasse überschrieben werden.

Parameter:

key (str) – Schlüssel des DataFrames (bei dict-Daten).
df (pd.DataFrame) – Eingabedaten.

Rückgabe:

Bereinigte Daten.

Rückgabetyp:

pd.DataFrame

save_data() → None¶: Speichert geladene Daten als CSV.

close() → None¶: Schließt WebDriver und löscht temporäre Ordner.

_load_config() → None¶

Lädt Crawler-spezifische Konfiguration aus der zentralen config.yaml.

Diese Methode liest:

Zugangsdaten (user, password, token, …)
URLs (für den jeweiligen Crawler)

Verursacht:

FileNotFoundError – Wenn keine gültige config.yaml gefunden wurde.
KeyError – Wenn keine passenden Einträge für diesen Crawler vorhanden sind.

__enter__() → WebCrawler¶

Context-Manager-Einstiegspunkt.

Wird automatisch aufgerufen, wenn der Crawler in einem with-Block verwendet wird. Gibt die aktuelle Instanz zurück, sodass alle Methoden wie gewohnt verfügbar sind.

Beispiel:

>>> with WebCrawler(browser="edge", headless=True) as crawler:
...     crawler.login()
...     crawler.download_data()
...     crawler.process_data()
...     crawler.save_data()
# Nach Ende des Blocks wird automatisch close() ausgeführt.

__exit__(exc_type, exc_value, traceback) → bool¶

Context-Manager-Ausstiegspunkt.

Wird automatisch aufgerufen, wenn der with-Block endet, unabhängig davon, ob ein Fehler aufgetreten ist. Führt close() aus, um den WebDriver zu schließen und temporäre Dateien zu löschen.

Parameter:

exc_type – Typ der Exception (falls eine aufgetreten ist)
exc_value – Exception-Instanz
traceback – Traceback-Objekt der Exception

Rückgabe:

False, damit Exceptions im with-Block nicht unterdrückt werden. (Python wirft sie weiter.)

Rückgabetyp:

bool

wait_for_element(by: str, selector: str, timeout: int = 15) → selenium.webdriver.remote.webelement.WebElement¶

Wartet auf das Vorhandensein eines Elements und gibt es zurück.

Parameter:

by (str | selenium.webdriver.common.by.By) – Suchstrategie für Selenium. Akzeptiert entweder: - einen case‑insensitiven String-Key (siehe Accepted string keys) - oder direkt eine Selenium-By‑Konstante (z. B. By.CSS_SELECTOR). Bei Übergabe eines bereits aufgelösten By/Tuple wird dieses direkt verwendet.
selector (str) – Selektor-String passend zur gewählten Strategie (z. B. CSS-Selector oder XPath).
timeout (int, optional) – Maximale Wartezeit in Sekunden. Standard ist 15.

Rückgabe:

Das gefundene Webelement.

Rückgabetyp:

WebElement

Verursacht:

selenium.common.exceptions.TimeoutException – Wenn das Element innerhalb der timeout-Zeit nicht gefunden wird.

Accepted string keys (case-insensitive) and mapping:

„id“ -> By.ID
„name“ -> By.NAME
„css“, „css selector“-> By.CSS_SELECTOR
„xpath“ -> By.XPATH
„link text“ -> By.LINK_TEXT
„partial link text“ -> By.PARTIAL_LINK_TEXT
„tag“ -> By.TAG_NAME
„class“ -> By.CLASS_NAME

Default behavior:

Wenn ein unbekannter String-Key übergeben wird, wird By.CSS_SELECTOR als Fallback verwendet. Wenn bereits eine By-Konstante oder ein Tuple (By.SOMETHING, selector) übergeben wird, wird dieser Wert unverändert verwendet.

Examples

>>> # mit String-Key (case-insensitive)
>>> elem = self.wait_for_element("css", "div.my-class")
>>> # mit vollständigem Key
>>> elem = self.wait_for_element("css selector", "div.my-class")
>>> # mit Selenium By-Konstante
>>> from selenium.webdriver.common.by import By
>>> elem = self.wait_for_element(By.ID, "username")
>>> # direktes Tuple (By, selector) möglich, falls intern verwendet
>>> elem = self.wait_for_element((By.XPATH, "//button[text()=\"OK\"]"), None)

wait_clickable_and_click(by: str, selector: str, timeout: int = 15) → None¶

Wartet auf ein Element und klickt es dann an.

Parameter:

by (str | By) – Suchstrategie oder By-Konstante.
selector (str) – Selektor-String.
timeout (int, optional) – Timeout in Sekunden. Standard 15.

Siehe auch

wait_for_element: Wartet auf das Element und gibt es zurück (verwendet von dieser Methode).

Sphinx cross-reference (für generierte Docs / IDE‑Plugins):: wait_for_element() or fully qualified: wait_for_element()

find_first_matching_element(selectors: list[tuple[str, str]], timeout_each: int = 10, debug_msg: bool = False) → selenium.webdriver.remote.webelement.WebElement¶

Versucht mehrere Selektoren nacheinander und gibt das erste gefundene Element zurück. :param selectors: Liste von (by, selector)-Tupeln. :param timeout_each: Timeout pro Selektor in Sekunden.

Rückgabe:: Erstes gefundenes Element.
Rückgabetyp:: WebElement
Verursacht:: selenium.common.exceptions.TimeoutException – Wenn kein Element für die gegebenen Selektoren gefunden wird.

Example

>>> selectors = (("css", "div.class1"), ("xpath", "//div[@id='main']"))
>>> elem = self.find_first_matching_element(selectors, timeout_each=5)

See also: wait_for_element: Wartet auf das Element und gibt es zurück (verwendet von dieser Methode).

Sphinx cross-reference (für generierte Docs / IDE‑Plugins):: wait_for_element() or fully qualified: wait_for_element()

click_first_matching_element(selectors: list[tuple[str, str]], timeout_each: int = 10) → None¶

Versucht mehrere Selektoren nacheinander und klickt das erste gefundene Element an. :param selectors: Liste von (by, selector)-Tupeln. :param timeout_each: Timeout pro Selektor in Sekunden.

Verursacht:: selenium.common.exceptions.TimeoutException – Wenn kein Element für die gegebenen Selektoren gefunden wird.

Example

>>> selectors = (("css", "button.accept"), ("xpath", "//button[text()='Accept']"))
>>> self.click_first_matching_element(selectors, timeout_each=5)

find_all_in(elem: selenium.webdriver.remote.webelement.WebElement, selectors: list[tuple[str, str]], debug_msg: bool = False) → list[selenium.webdriver.remote.webelement.WebElement]¶: Findet alle passenden Unterelemente innerhalb eines Elements.

find_first_in(elem: selenium.webdriver.remote.webelement.WebElement, selectors: list[tuple[str, str]], debug_msg: bool = False) → selenium.webdriver.remote.webelement.WebElement¶: Findet das erste passende Unterelement innerhalb eines Elements.

scroll_into_view(element) → None¶: Scrollt ein Element ins Viewport.

click_js(element) → None¶: Klickt ein Element via JavaScript (Fallback bei Overlay o.Ä.).

accept_cookies_if_present(selectors: tuple[str, Ellipsis] = ('button#onetrust-accept-btn-handler', "button[aria-label='Akzeptieren']", 'button.cookie-accept'), timeout_each: int = 3) → bool¶

Versucht gängige Cookie-Banner wegzuklicken.

Parameter:

selectors – Liste möglicher CSS-Selektoren für Zustimmungs-Buttons.
timeout_each – Zeitfenster pro Selektor.

Rückgabe:

True, wenn ein Banner geschlossen wurde.

Rückgabetyp:

bool

_wait_for_new_file(timeout: float = 30, check_interval: float = 0.5, include_temp: bool = True) → str | None¶

Wartet auf eine neue Datei im Download-Ordner und gibt deren Dateinamen zurück.

Parameter:

timeout – Maximale Wartezeit in Sekunden.
check_interval – Prüfintervall in Sekunden.
include_temp – Ob temporäre Dateien (.crdownload/.tmp) berücksichtigt werden.

Rückgabe:

Der Dateiname der neu erkannten Datei oder None bei Timeout.

_read_temp_files(sep: str = ';', max_retries: int = 10, retry_wait: float = 1.0, check_interval: float = 0.1, download_timeout: float = 10.0) → bool¶

Liest Dateien aus dem Download-Ordner in self.data.

Unterstützt CSV, XLS, XLSX. Wartet optional, bis temporäre Download-Dateien (.crdownload/.tmp) verschwunden sind.

Rückgabe:: True bei Erfolg, sonst False.

_retry_func(func, max_retries: int = 3, wait_seconds: float = 1.0, args: tuple | None = None, kwargs: dict | None = None) → bool¶

Versucht die Funktion mehrfach bei Fehlschlag.

Parameter:

func – Funktion, die ausgeführt werden soll.
max_retries – Maximale Anzahl an Versuchen.
wait_seconds – Wartezeit zwischen den Versuchen.

Rückgabe:

True bei erfolgreicher Ausfürhung, sonst False.

Rückgabetyp:

bool

_wait_for_manual_exit(msg: str = None)¶

Wartet auf manuelles Schließen des Browsers durch den Nutzer.

Parameter:

msg – Nachricht, die angezeigt werden soll. (Optional)

Rückgabe:

_wait_for_condition(condition_func, timeout: float = 30.0, check_interval: float = 0.5) → bool¶

Wartet, bis eine Bedingungsfunktion True zurückgibt.

Parameter:

condition_func – Funktion, die eine boolesche Bedingung prüft.
timeout – Maximale Wartezeit in Sekunden.
check_interval – Prüfintervall in Sekunden.

Rückgabe:

True, wenn die Bedingung erfüllt wurde, sonst False.

Rückgabetyp:

bool

_delete_header(df: pandas.DataFrame, header_key: str = 'datum') → pandas.DataFrame¶

Löscht die Header-Zeile bis zum header_key aus einem DataFrame und setzt die Spaltennamen.

Parameter:

df – Eingabe-DataFrame.
header_key – Key der Spalte, die im Header enthalten sein muss. (Standard: ‚datum‘)

Rückgabe:

DataFrame ohne Header-Zeile.

_normalize_dataframe(df: pandas.DataFrame, remove_nan: bool = False, date_as_str: bool = False) → pandas.DataFrame¶

Normalisiert die Transaktionsdaten eines DataFrames. - Datumsspalten in einheitliches Format bringen - Betragsspalten bereinigen - Spaltennamen vereinheitlichen

Parameter:

df – Eingabe-DataFrame.
remove_nan – Ob Zeilen mit NaN-Werten entfernt werden sollen. (Standard: False)

Rückgabe:

DataFrame mit normalisierten Daten.

_normalize_date_in_dataframe(df: pandas.DataFrame, date_column: str, *, date_as_str: bool = False, dayfirst: bool = True) → pandas.DataFrame¶

Normalisiert die Datumswerte in der angegebenen Spalte eines DataFrames.

Parameter:

df – Eingabe-DataFrame.
date_column – Name der Spalte mit den Datumswerten.
date_as_str – Ob das Datum als String zurückgegeben werden soll. (Standard: False)
dayfirst – Ob der Tag vor dem Monat steht. (Standard: True)

Rückgabe:

DataFrame mit normalisierten Datumswerten.

_normalize_amount_in_dataframe(df: pandas.DataFrame, amount_column: str, *, remove_nan: bool = False) → pandas.DataFrame¶

Normalisiert die Beträge in der angegebenen Spalte eines DataFrames.

Parameter:

df – Eingabe-DataFrame.
amount_column – Name der Spalte mit den Beträgen.
remove_nan – Ob Zeilen mit NaN-Werten entfernt werden sollen. (Standard: False)

Rückgabe:

DataFrame mit normalisierten Beträgen.

_normalize_amount(value: Any) → float¶

Bereinigt Währungs-Strings und konvertiert sie in float. Unterstützt pandas Series, DataFrames und einzelne Strings.

Parameter:: value – Eingabewert (Series, DataFrame oder String).
Rückgabe:: Bereinigter Wert als float, Series oder DataFrame.

_filter_out_rows_by_needles(df: pandas.DataFrame, column: str, needles: list[str], *, case_sensitive: bool = False, allow_regex: bool = False, whole_word: bool = False, keep_na: bool = True) → pandas.DataFrame¶

Entfernt Zeilen, wenn column einen Begriff der needles enthält.

Parameter:

df – Eingabe-DataFrame.
column – Zu durchsuchende Spalte.
needles – Liste Suchbegriffe (oder Regex, wenn allow_regex=True).
case_sensitive – Groß-/Kleinschreibung beachten?
allow_regex – needles als echte Regex behandeln?
whole_word – Nur ganze Wörter matchen (setzt allow_regex=True intern).
keep_na – NaN in column behalten (True) oder als „kein Treffer“ behandeln (False).

Rückgabe:

Gefiltertes DataFrame (Treffer werden entfernt).

_filter_in_rows_by_needles(df: pandas.DataFrame, column: str, needles: list[str], *, case_sensitive: bool = False, allow_regex: bool = False, whole_word: bool = False, keep_na: bool = True) → pandas.DataFrame¶

Behalte nur Zeilen, wenn column einen Begriff der needles enthält. :param df: Eingabe-DataFrame. :param column: Zu durchsuchende Spalte. :param needles: Liste Suchbegriffe (oder Regex, wenn allow_regex=True). :param case_sensitive: Groß-/Kleinschreibung beachten? :param allow_regex: needles als echte Regex behandeln? :param whole_word: Nur ganze Wörter matchen (setzt allow_regex=True intern). :param keep_na: NaN in column behalten (True) oder als „kein Treffer“ behandeln (False).

Rückgabe:: Gefiltertes DataFrame (nur Treffer werden behalten).

_filter_columns_by_names(df: pandas.DataFrame, column_names: list[str], *, add_missing: bool = False, fill_value=pd.NA, case_insensitive: bool = False) → pandas.DataFrame¶

Behält nur die Spalten in column_names (in derselben Reihenfolge). Optional:

add_missing: fehlende Spalten erzeugen (mit fill_value)

case_insensitive: Spaltennamen case-insensitiv auflösen

_rename_columns_by_map(df: pandas.DataFrame, rename_map: dict[str, str], *, case_insensitive: bool = False)¶

Benennt Spalten gemäß rename_map um. :param df: Eingabe-DataFrame. :param rename_map: Dict mit {alter_spaltenname: neuer_spaltenname}. :param case_insensitive: Ob Spaltennamen case-insensitiv gesucht werden sollen. :type case_insensitive: bool, optional

Optional: case_insensitive: Sucht Spaltennamen case-insensitiv.

Rückgabe:: DataFrame mit umbenannten Spalten.

_abort_windows_passkey(tries: int = 10, timeout: int = 10) → bool¶: Versucht, einen nativen Windows-Passkey/Hello/WebAuthn-Dialog zu schließen. Priorität: pywinauto -> ctypes SendInput -> pyautogui/keyboard -> ESC an Browser. Gibt True zurück, wenn mind. ein Abbruchversuch gesendet wurde.

_log_error_with_debug_msg(msg: str | None = None) → None¶

Loggt eine Debug-Nachricht mit Funktionsname, Dateiname und Zeilennummer der aufrufenden Stelle (nicht der Logger-Funktion selbst).

Parameter:: msg – Zusätzliche Nachricht, die geloggt werden soll. (Optional)
Rückgabe:: None