diff --git a/brancheneinstufung.py b/brancheneinstufung.py index e4a8bb39..0eae4c5d 100644 --- a/brancheneinstufung.py +++ b/brancheneinstufung.py @@ -2421,210 +2421,148 @@ class GoogleSheetHandler: """ Initialisiert den Handler, stellt die Verbindung her und laedt die Daten. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist + # Holen Sie eine Logger-Instanz fuer diese Klasse + self.logger = logging.getLogger(__name__ + ".GoogleSheetHandler") # Initialisieren Sie die Attribute self.sheet = None - # Daten werden hier als Instanzvariable gespeichert, um nicht bei jedem Zugriff neu laden zu muessen - self.sheet_values = [] - # header_rows sind fix, aber wir koennen sie hier zur Klarheit definieren - self._header_rows = 5 # Annahme: Die ersten 5 Zeilen sind Header - - logger.info("Initialisiere GoogleSheetHandler...") + # ... + self.logger.info("Initialisiere GoogleSheetHandler...") try: # Verbindung wird bei der Initialisierung aufgebaut - # Der _connect Aufruf ist mit retry_on_failure dekoriert. - # Wenn _connect eine Exception wirft (auch nach Retries), wird diese hier gefangen. self._connect() # Daten werden ebenfalls bei der Initialisierung geladen, nur wenn die Verbindung erfolgreich war if self.sheet: - # Der load_data Aufruf ist mit retry_on_failure dekoriert. - # Wenn load_data eine Exception wirft (auch nach Retries), wird diese hier gefangen. - self.load_data() # Erste Datenladung nach erfolgreicher Verbindung + self.load_data() # Erste Datenladung nach erfolgreicher Verbindung else: - # Wenn die Verbindung fehlschlug (sheet ist None), aber keine Exception geworfen wurde (sollte nicht passieren) - logger.critical("GoogleSheetHandler Init FEHLER: Verbindung konnte nicht hergestellt werden (sheet ist None).") - # Werfen Sie eine spezifische Exception - raise ConnectionError("Google Sheet Handler Init failed: Verbindung konnte nicht hergestellt werden.") - + # Wenn die Verbindung fehlschlug (sheet ist None), aber keine Exception geworfen wurde (sollte nicht passieren) + self.logger.critical( + "GoogleSheetHandler Init FEHLER: Verbindung konnte nicht hergestellt werden (sheet ist None)." + ) + raise ConnectionError( + "Google Sheet Handler Init failed: Verbindung konnte nicht hergestellt werden." + ) except Exception as e: # Fehler bei der Initialisierung (entweder von _connect oder load_data nach Retries) - # werden hier gefangen und erneut geworfen, damit die main-Funktion (Block 34) entsprechend reagieren kann. - logger.critical(f"FATAL: Fehler bei Initialisierung von GoogleSheetHandler: {type(e).__name__} - {e}") - # Loggen Sie den Traceback fuer detailliertere Fehlerinformationen - logger.debug(traceback.format_exc()) - # Werfen Sie eine aussagekraeftige Exception - raise ConnectionError(f"Google Sheet Handler Init failed: {e}") # Signalisiert Verbindungsproblem + self.logger.critical( + f"FATAL: Fehler bei Initialisierung von GoogleSheetHandler: {type(e).__name__} - {e}" + ) + self.logger.debug(traceback.format_exc()) + raise ConnectionError(f"Google Sheet Handler Init failed: {e}") - - @retry_on_failure # Wende den Decorator an, da es externe Calls macht + @retry_on_failure def _connect(self): """Stellt Verbindung zum Google Sheet her.""" - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - self.sheet = None # Setze sheet vor dem Versuch auf None - logger.info("Versuche Verbindung mit Google Sheets herstellen...") + self.sheet = None + self.logger.info("Versuche Verbindung mit Google Sheets herstellen...") try: - # Stellen Sie sicher, dass CREDENTIALS_FILE korrekt ist if not os.path.exists(CREDENTIALS_FILE): - # Werfen Sie FileNotFoundError, dies wird vom retry_on_failure als permanent behandelt. - raise FileNotFoundError(f"Credential-Datei nicht gefunden: {CREDENTIALS_FILE}") + raise FileNotFoundError(f"Credential-Datei nicht gefunden: {CREDENTIALS_FILE}") - # Definieren Sie den Scope fuer den Zugriff auf Google Sheets scope = ["https://www.googleapis.com/auth/spreadsheets"] - # Laden Sie die Credentials aus der JSON-Datei creds = ServiceAccountCredentials.from_json_keyfile_name(CREDENTIALS_FILE, scope) - # Autorisiere die Verbindung mit den Credentials gc = gspread.authorize(creds) - # Oeffne das Sheet ueber die URL aus Config - # Dieser Aufruf kann gspread.exceptions.SpreadsheetNotFound werfen, was vom Decorator behandelt wird. - # Er kann auch andere gspread.exceptions.APIError oder requests.exceptions.RequestException werfen. - sh = gc.open_by_url(Config.SHEET_URL) # Nutzt die URL aus Config (Block 1) - - # Greife auf das erste Blatt zu (Index 0, ueblicherweise "Tabelle1") - # Dieser Aufruf kann gspread.exceptions.WorksheetNotFound werfen, wenn das erste Blatt fehlt. - # Wenn das erste Blatt fehlt, ist das ein Konfigurationsproblem und sollte eine Exception werfen. - self.sheet = sh.sheet1 # Greift auf das erste Blatt zu (Index 0) - - logger.info("Verbindung zu Google Sheets erfolgreich.") - # Die Methode gibt implizit None zurueck, wenn keine Exception geworfen wird. - - - # Spezifische Fehlerbehandlung fuer gspread/requests Fehler, die vom Decorator behandelt werden. - # Wenn hier eine Exception durchkommt, hat der Decorator (nach Retries oder weil permanent) aufgegeben. + sh = gc.open_by_url(Config.SHEET_URL) + self.sheet = sh.sheet1 + self.logger.info("Verbindung zu Google Sheets erfolgreich.") except (gspread.exceptions.APIError, requests.exceptions.RequestException, FileNotFoundError) as e: - # Der Decorator hat diesen Fehler bereits geloggt und (falls nicht permanent) Retries versucht. - # Werfen Sie den Fehler erneut, damit der Aufrufer (die __init__ Methode) ihn fangen kann. - raise e - + raise e except Exception as e: - # Fangen Sie andere unerwartete Verbindungsfehler ab (z.B. Probleme mit der Bibliothek selbst) - logger.error(f"FEHLER bei der Google Sheets Verbindung: {type(e).__name__} - {e}") - # Loggen Sie den Traceback fuer detaillierte Fehlerinformationen - logger.debug(traceback.format_exc()) - # Werfen Sie die Exception erneut, damit der Aufrufer sie behandeln kann - raise e + self.logger.error(f"FEHLER bei der Google Sheets Verbindung: {type(e).__name__} - {e}") + self.logger.debug(traceback.format_exc()) + raise e - - @retry_on_failure # Wende den Decorator an, da es externe Calls macht + @retry_on_failure def load_data(self): """ Laedt alle Daten aus dem Sheet und aktualisiert self.sheet_values. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefen Sie, ob eine Sheet-Verbindung vorhanden ist if not self.sheet: - logger.error("Fehler: Keine Sheet-Verbindung zum Laden der Daten.") - self.sheet_values = [] # Stelle sicher, dass die Datenliste leer ist - return False # Signalisiert Fehler beim Laden + self.logger.error("Fehler: Keine Sheet-Verbindung zum Laden der Daten.") + self.sheet_values = [] + return False - logger.info("Lade Daten aus Google Sheet...") + self.logger.info("Lade Daten aus Google Sheet...") try: - # Lade alle Werte aus dem Sheet. Dieser Aufruf kann Exceptions werfen. - # get_all_values() laedt standardmaessig das erste Blatt ("Tabelle1"). - # Wenn das Blatt umbenannt wurde oder fehlt, kann dies einen Fehler werfen. self.sheet_values = self.sheet.get_all_values() - # Wenn get_all_values eine leere Liste zurueckgibt (z.B. Sheet ist leer) if not self.sheet_values: - logger.warning("Google Sheet scheint leer zu sein oder get_all_values() lieferte keine Daten.") - # Wenn die erste Zeile nicht geladen werden kann (z.B. leeres Sheet), headers ist leer - self.headers = [] # Setzen Sie die Header-Liste auf leer - return True # Ladevorgang war technisch erfolgreich (keine Exception), aber keine Daten + self.logger.warning( + "Google Sheet scheint leer zu sein oder get_all_values() lieferte keine Daten." + ) + self.headers = [] + return True - - # Logge die Anzahl der Zeilen und Spalten, die geladen wurden num_rows = len(self.sheet_values) num_cols = len(self.sheet_values[0]) if num_rows > 0 else 0 - logger.info(f"Daten neu geladen: {num_rows} Zeilen, {num_cols} Spalten.") + self.logger.info(f"Daten neu geladen: {num_rows} Zeilen, {num_cols} Spalten.") - # Optional: Ueberpruefen Sie, ob die Anzahl der Spalten mindestens dem hoechsten Index in COLUMN_MAP entspricht try: - # Finden Sie den hoechsten Index in COLUMN_MAP (Block 1) - max_col_idx_in_map = max(COLUMN_MAP.values()) - # Ueberpruefen Sie, ob die Anzahl der geladenen Spalten ausreicht - if num_cols <= max_col_idx_in_map: # Verwenden Sie <= weil Indizes 0-basiert sind - # Logge eine Warnung, wenn das Mapping auf Spalten zeigt, die nicht geladen wurden - logger.warning(f"Geladenes Sheet hat {num_cols} Spalten, erwartet werden aber mindestens {max_col_idx_in_map + 1} basierend auf COLUMN_MAP. Das COLUMN_MAP passt moeglicherweise nicht zum Sheet!") - except ValueError: # Tritt auf, wenn COLUMN_MAP leer ist - logger.warning("COLUMN_MAP scheint leer zu sein oder enthaelt keine Werte. Kann Spaltenanzahl nicht pruefen.") + max_col_idx_in_map = max(COLUMN_MAP.values()) + if num_cols <= max_col_idx_in_map: + self.logger.warning( + f"Geladenes Sheet hat {num_cols} Spalten, erwartet werden aber mindestens " + f"{max_col_idx_in_map + 1} basierend auf COLUMN_MAP." + ) + except ValueError: + self.logger.warning( + "COLUMN_MAP scheint leer zu sein oder enthaelt keine Werte. Kann Spaltenanzahl nicht pruefen." + ) except Exception as e: - # Fangen Sie andere unerwartete Fehler bei der Pruefung ab - logger.error(f"Fehler bei der Pruefung der Spaltenanzahl gegen COLUMN_MAP: {e}") + self.logger.error(f"Fehler bei der Pruefung der Spaltenanzahl gegen COLUMN_MAP: {e}") - - # Speichere die erste Zeile als Header-Namen (optional, kann fuer spaetere Zuordnung nuetzlich sein) - # Dies sollte erst nach Pruefung auf leere sheet_values geschehen if num_rows > 0: - self.headers = self.sheet_values[0] + self.headers = self.sheet_values[0] else: - self.headers = [] + self.headers = [] - - return True # Signalisiert erfolgreiches Laden (auch wenn keine Daten da sind) - - # Spezifische Fehlerbehandlung fuer gspread/requests Fehler, die vom Decorator behandelt werden. - # Wenn hier eine Exception durchkommt, hat der Decorator (nach Retries) aufgegeben. + return True except (gspread.exceptions.APIError, requests.exceptions.RequestException) as e: - # Der Decorator hat diesen Fehler bereits geloggt und Retries versucht. - # Werfen Sie den Fehler erneut, damit der Aufrufer (die __init__ Methode) ihn fangen kann. - raise e - + raise e except Exception as e: - # Fangen Sie andere unerwartete Ladefehler ab - logger.error(f"Allgemeiner Fehler beim Laden der Google Sheet Daten: {type(e).__name__} - {e}") - # Loggen Sie den Traceback - logger.debug(traceback.format_exc()) - # Werfen Sie die Exception erneut, damit der Aufrufer sie behandeln kann - raise e - + self.logger.error( + f"Allgemeiner Fehler beim Laden der Google Sheet Daten: {type(e).__name__} - {e}" + ) + self.logger.debug(traceback.format_exc()) + raise e def get_data(self): """ Gibt die aktuell im Handler gespeicherten Datenzeilen zurueck (ohne die ersten N Header-Zeilen). """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefen Sie, ob Daten geladen wurden und ob genuegend Zeilen fuer Header vorhanden sind if not self.sheet_values or len(self.sheet_values) <= self._header_rows: - # Logge nur auf Debug, da dies oft passiert, wenn das Sheet leer ist oder nur Header enthaelt - logger.debug(f"get_data: Keine Datenzeilen verfuegbar (geladen: {len(self.sheet_values) if self.sheet_values else 0} Zeilen, {self._header_rows} Header).") - return [] # Gebe eine leere Liste zurueck, wenn keine Datenzeilen vorhanden sind - - # Gibt eine Slice der Liste zurueck (Kopie, um unbeabsichtigte Aenderungen am Original zu vermeiden) + self.logger.debug( + f"get_data: Keine Datenzeilen verfuegbar " + f"(geladen: {len(self.sheet_values) if self.sheet_values else 0} Zeilen, " + f"{self._header_rows} Header)." + ) + return [] return self.sheet_values[self._header_rows:].copy() - def get_all_data_with_headers(self): - """Gibt alle aktuell im Handler gespeicherten Daten inklusive Header zurueck.""" - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefen Sie, ob Daten geladen wurden - if not self.sheet_values: - logger.debug("get_all_data_with_headers: Keine Daten im Handler gespeichert.") - return [] # Gebe eine leere Liste zurueck, wenn keine Daten geladen wurden - - # Geben Sie eine Kopie der gesamten Datenliste zurueck - return self.sheet_values.copy() - + """Gibt alle aktuell im Handler gespeicherten Daten inklusive Header zurueck.""" + if not self.sheet_values: + self.logger.debug("get_all_data_with_headers: Keine Daten im Handler gespeichert.") + return [] + return self.sheet_values.copy() def _get_col_letter(self, col_idx_1_based): """ Konvertiert einen 1-basierten Spaltenindex in den entsprechenden Google Sheets Spaltenbuchstaben (A, B, ..., Z, AA, ...). """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefen Sie, ob der Index ein gueltiger Integer und positiv ist if not isinstance(col_idx_1_based, int) or col_idx_1_based < 1: - # Logge den Fehler auf Error-Level - logger.error(f"Ungueltiger Spaltenindex ({col_idx_1_based}) fuer _get_col_letter erhalten.") - return None # Gebe None zurueck bei ungueltigem Index + self.logger.error( + f"Ungueltiger Spaltenindex ({col_idx_1_based}) fuer _get_col_letter erhalten." + ) + return None - # Implementierung zur Konvertierung von Zahl zu Buchstaben string = "" n = col_idx_1_based while n > 0: n, remainder = divmod(n - 1, 26) string = chr(65 + remainder) + string - return string # Gebe den berechneten Spaltenbuchstaben zurueck - + return string def get_start_row_index(self, check_column_key, min_sheet_row=7): """ @@ -2639,90 +2577,84 @@ class GoogleSheetHandler: Standardmaessig ab Zeile 7 (erste Zeile nach 5 Headern und einer leeren). Returns: - int: Der 0-basierte Index in der Datenliste (ohne Header) der ersten Zeile + int: Der 0-basierten Index in der Datenliste (ohne Header) der ersten Zeile mit einem EXAKT leeren Wert in der Zielspalte innerhalb des Suchbereichs. Gibt -1 zurueck bei schwerwiegenden Fehlern (z.B. Schluessel fehlt in COLUMN_MAP). Gibt die Laenge der Datenliste zurueck, wenn keine leere Zelle im Suchbereich gefunden wurde. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Daten neu laden, um sicherzustellen, dass sie aktuell sind if not self.load_data(): - logger.error("Fehler beim Laden der Daten fuer get_start_row_index.") - return -1 # Signalisiert Fehler beim Laden + self.logger.error("Fehler beim Laden der Daten fuer get_start_row_index.") + return -1 - data_rows = self.get_data() # Datenzeilen ohne Header (Kopie) - # Wenn keine Datenzeilen vorhanden sind + data_rows = self.get_data() if not data_rows: - logger.info("Keine Datenzeilen im Sheet gefunden. Startindex fuer leere Zelle ist 0.") - return 0 # Die erste Datenzeile (Index 0) waere der Start + self.logger.info("Keine Datenzeilen im Sheet gefunden. Startindex fuer leere Zelle ist 0.") + return 0 - - # Ermitteln Sie den Index der zu pruefenden Spalte aus COLUMN_MAP (Block 1) check_column_index = COLUMN_MAP.get(check_column_key) if check_column_index is None: - logger.critical(f"FEHLER: Schluessel '{check_column_key}' nicht in COLUMN_MAP gefunden fuer get_start_row_index!") - return -1 # Signalisiert Fehler (Schluessel fehlt) + self.logger.critical( + f"FEHLER: Schluessel '{check_column_key}' nicht in COLUMN_MAP gefunden fuer get_start_row_index!" + ) + return -1 - # Ermitteln Sie den Spaltenbuchstaben fuer Logging actual_col_letter = self._get_col_letter(check_column_index + 1) - # Wenn Spaltenbuchstabe nicht ermittelt werden konnte (ungueltiger Index), loggen wir es. if actual_col_letter is None: - logger.error(f"FEHLER: Konnte Spaltenbuchstaben fuer Index {check_column_index + 1} nicht ermitteln.") - # Gehen Sie weiter, aber die Logs werden weniger informativ sein. - actual_col_letter = f"Index_{check_column_index + 1}" + self.logger.error( + f"FEHLER: Konnte Spaltenbuchstaben fuer Index {check_column_index + 1} nicht ermitteln." + ) + actual_col_letter = f"Index_{check_column_index + 1}" - - # Berechne den Startindex in der 0-basierten 'data_rows' Liste - # min_sheet_row (1-basiert) -> 0-basierten Index in all_data -> 0-basierten Index in data_rows - # 1-basierte Sheet-Zeilennummer minus 1 ergibt den 0-basierten Index in der Gesamtliste (all_data) - # Den Index der Header-Zeilen subtrahieren ergibt den 0-basierten Index in der Datenliste (data_rows) search_start_index_in_data = max(0, (min_sheet_row - 1) - self._header_rows) + self.logger.info( + f"get_start_row_index: Suche ab Daten-Index {search_start_index_in_data} " + f"(Sheet-Zeile {search_start_index_in_data + self._header_rows + 1}) " + f"nach EXAKT LEEREM Wert (=='') in Spalte '{check_column_key}' ({actual_col_letter})..." + ) - logger.info(f"get_start_row_index: Suche ab Daten-Index {search_start_index_in_data} (Sheet-Zeile {search_start_index_in_data + self._header_rows + 1}) nach EXAKT LEEREM Wert (=='') in Spalte '{check_column_key}' ({actual_col_letter})...") - - # Pruefen Sie, ob der Start-Suchindex ausserhalb der Datenliste liegt if search_start_index_in_data >= len(data_rows): - # Wenn der Startindex groesser oder gleich der Laenge der Datenliste ist, - # bedeutet dies, dass der gesamte Suchbereich ausserhalb der vorhandenen Daten liegt. - logger.warning(f"Start-Suchindex in Daten ({search_start_index_in_data}) liegt hinter der letzten Datenzeile ({len(data_rows)}). Keine leere Zelle gefunden im Suchbereich.") - # Rueckgabe der Laenge der Datenliste signalisiert, dass keine leere Zelle im Suchbereich gefunden wurde. + self.logger.warning( + f"Start-Suchindex in Daten ({search_start_index_in_data}) liegt hinter der letzten Datenzeile ({len(data_rows)}). Keine leere Zelle gefunden im Suchbereich." + ) return len(data_rows) - # Iteriere ueber die Datenzeilen ab dem berechneten Startindex for i in range(search_start_index_in_data, len(data_rows)): - row = data_rows[i] # Die aktuelle Zeile in der Datenliste (0-basiert) - current_sheet_row = i + self._header_rows + 1 # 1-basierte Sheet-Zeilennummer fuer Logging + row = data_rows[i] + current_sheet_row = i + self._header_rows + 1 - cell_value = ""; is_exactly_empty = True - # Ueberpruefe, ob die Zeile lang genug ist, um auf die Spalte zuzugreifen + cell_value = "" + is_exactly_empty = True if len(row) > check_column_index: - cell_value = row[check_column_index] # Hole den Wert in der Zielspalte - # Pruefen Sie, ob der Wert EXAKT ein leerer String ist - if cell_value != "": is_exactly_empty = False + cell_value = row[check_column_index] + if cell_value != "": + is_exactly_empty = False else: - # Wenn die Zeile nicht lang genug ist, gilt die Zelle in der Zielspalte als leer - is_exactly_empty = True + is_exactly_empty = True - # Logge die ersten paar Zeilen, jede 1000. Zeile oder wenn eine leere Zelle gefunden wird log_debug = (i < search_start_index_in_data + 5) or (i % 1000 == 0) or is_exactly_empty if log_debug: - # Logge den Wert und den Status der Pruefung - logger.debug(f" -> Pruefe Daten-Index {i} (Sheet {current_sheet_row}): Wert in {actual_col_letter}='{str(cell_value).strip()}' (Roh='{cell_value}' Typ: {type(cell_value)}). Ist exakt leer ('')? {is_exactly_empty}") + self.logger.debug( + f" -> Pruefe Daten-Index {i} (Sheet {current_sheet_row}): " + f"Wert in {actual_col_letter}='{str(cell_value).strip()}' " + f"(Roh='{cell_value}' Typ: {type(cell_value)}). Leer? {is_exactly_empty}" + ) - # Wenn eine exakt leere Zelle gefunden wurde if is_exactly_empty: - logger.info(f"Erste Zeile ab Sheet-Zeile {min_sheet_row} mit EXAKT LEEREM Wert in Spalte {actual_col_letter} gefunden: Zeile {current_sheet_row} (Daten-Index {i})") - return i # Gebe den 0-basierten Index in der Datenliste zurueck (dies ist der Startindex fuer die Verarbeitung) + self.logger.info( + f"Erste Zeile ab Sheet-Zeile {min_sheet_row} mit EXAKT LEEREM Wert in Spalte " + f"{actual_col_letter} gefunden: Zeile {current_sheet_row} (Daten-Index {i})" + ) + return i - # Wenn die Schleife durchlaeuft, ohne eine exakt leere Zelle im Suchbereich zu finden - last_data_index = len(data_rows) # Der Index nach der letzten Datenzeile - logger.info(f"Alle Zeilen ab Daten-Index {search_start_index_in_data} im Suchbereich haben einen nicht-leeren Wert in Spalte {actual_col_letter}. Naechster Daten-Index waere {last_data_index}.") - # Rueckgabe der Laenge der Datenliste signalisiert, dass keine leere Zelle gefunden wurde. + last_data_index = len(data_rows) + self.logger.info( + f"Alle Zeilen ab Daten-Index {search_start_index_in_data} im Suchbereich haben einen " + f"nicht-leeren Wert in Spalte {actual_col_letter}. Naechster Daten-Index waere {last_data_index}." + ) return last_data_index - - @retry_on_failure # Wende den Decorator an, da es externe Calls macht + @retry_on_failure def batch_update_cells(self, update_data): """ Fuehrt ein Batch-Update im Google Sheet durch. Beinhaltet robustere @@ -2736,45 +2668,28 @@ class GoogleSheetHandler: Returns: bool: True bei Erfolg (nach allen Retries), False bei endgueltigem Fehler. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefen Sie, ob eine Sheet-Verbindung vorhanden ist if not self.sheet: - logger.error("FEHLER: Keine Sheet-Verbindung fuer Batch-Update.") - return False # Signalisiert Fehler + self.logger.error("FEHLER: Keine Sheet-Verbindung fuer Batch-Update.") + return False - # Wenn keine Update-Daten vorhanden sind if not update_data: - # logger.debug("Keine Daten fuer Batch-Update vorhanden.") # Zu viel Laerm im Debug - return True # Nichts zu tun ist technisch ein Erfolg - - - # Die retry_on_failure Logik kuemmert sich um die Wiederholung und das Werfen - # der Exception im Fehlerfall. Wir muessen hier nur den Aufruf machen und - # das Ergebnis (oder die Exception) weitergeben. + return True try: - # Schaetze die Anzahl der zu aktualisierenden Zellen fuer Logging - total_cells_to_update = sum(len(row) for item in update_data for row in item.get('values', [])) - logger.debug(f" -> Versuche sheet.batch_update mit {len(update_data)} Anfragen ({total_cells_to_update} Zellen)...") - - # Fuehren Sie das Batch-Update durch. gspread.batch_update wirft bei Fehlern Exceptions, - # die vom @retry_on_failure Decorator gefangen werden. - # value_input_option='USER_ENTERED' interpretiert die Eingaben wie ein Nutzer. + total_cells_to_update = sum( + len(row) for item in update_data for row in item.get('values', []) + ) + self.logger.debug( + f" -> Versuche sheet.batch_update mit {len(update_data)} Anfragen " + f"({total_cells_to_update} Zellen)..." + ) self.sheet.batch_update(update_data, value_input_option='USER_ENTERED') - - # Wenn keine Exception aufgetreten ist, war der Aufruf (ggf. nach Retries) erfolgreich. - # logger.debug(f" -> sheet.batch_update erfolgreich abgeschlossen.") # Zu viel Laerm im Debug - return True # Signalisiert Erfolg - - # Exceptions werden vom retry_on_failure gefangen und (im Fehlerfall) neu geworfen. - # Wenn eine Exception hier durchkommt, hat retry_on_failure aufgegeben. - except Exception as e: - # Der endgueltige Fehler wurde bereits vom Decorator geloggt. - # Wir fangen ihn hier nur, um False zurueckzugeben, wie in der Signatur versprochen. - # Das re-raising im Decorator sorgt dafuer, dass wir hier landen, wenn der Decorator aufgibt. - logger.error(f"Endgueltiger Fehler beim Batch-Update nach Retries. Kann {len(update_data)} Operationen nicht durchfuehren.") - # Der Traceback wurde bereits vom Decorator (im except Exception Fall) geloggt. - return False # Signalisiert endgueltigen Fehler + return True + except Exception: + self.logger.error( + f"Endgueltiger Fehler beim Batch-Update nach Retries. Kann {len(update_data)} Operationen nicht durchfuehren." + ) + return False # --- WIKIPEDIA SCRAPER CLASS --- @@ -2802,741 +2717,483 @@ class WikipediaScraper: self.logger.debug("WikipediaScraper initialisiert.") # User-Agent fuer Requests (nutzt Config, Fallback wenn nicht gesetzt) - self.user_agent = user_agent or getattr(Config, 'USER_AGENT', 'Mozilla/5.0 (compatible; UnternehmenSkript/1.0; +http://www.example.com/bot)') # Beispiel URL anpassen - self.session = requests.Session() # Nutzt eine Requests Session fuer bessere Performance bei mehreren Anfragen - self.session.headers.update({'User-Agent': self.user_agent}) # Setzt den User-Agent Header + self.user_agent = user_agent or getattr( + Config, 'USER_AGENT', + 'Mozilla/5.0 (compatible; UnternehmenSkript/1.0; +http://www.example.com/bot)' + ) + self.session = requests.Session() + self.session.headers.update({'User-Agent': self.user_agent}) self.logger.debug(f"Requests Session mit User-Agent '{self.user_agent}' initialisiert.") # Keywords fuer die Infobox-Extraktion self.keywords_map = { - 'branche': ['branche', 'wirtschaftszweig', 'industry', 'taetigkeit', 'sektor', 'produkte', 'leistungen'], # Umlaute vermeiden - 'umsatz': ['umsatz', 'erloes', 'revenue', 'jahresumsatz', 'konzernumsatz', 'ergebnis'], # Umlaute vermeiden - 'mitarbeiter': ['mitarbeiter', 'mitarbeiterzahl', 'beschaeftigte', 'employees', 'number of employees', 'personal', 'belegschaft'] # Umlaute vermeiden + 'branche': ['branche', 'wirtschaftszweig', 'industry', 'taetigkeit', 'sektor', 'produkte', 'leistungen'], + 'umsatz': ['umsatz', 'erloes', 'revenue', 'jahresumsatz', 'konzernumsatz', 'ergebnis'], + 'mitarbeiter': ['mitarbeiter', 'mitarbeiterzahl', 'beschaeftigte', 'employees', 'number of employees', 'personal', 'belegschaft'] } # Konfiguriere die wikipedia-Bibliothek try: - wiki_lang = getattr(Config, 'LANG', 'de') # Sprache aus Config holen - wikipedia.set_lang(wiki_lang) # Setzt die Sprache fuer die wikipedia Bibliothek - # Aktivieren Sie Rate Limiting, um die Wikipedia-API nicht zu ueberlasten - wikipedia.set_rate_limiting(True, min_wait=0.1) # Minimum 0.1 Sekunden warten zwischen API calls - self.logger.info(f"Wikipedia library language set to '{wiki_lang}'. Rate limiting enabled (min_wait=0.1).") + wiki_lang = getattr(Config, 'LANG', 'de') + wikipedia.set_lang(wiki_lang) + wikipedia.set_rate_limiting(True, min_wait=0.1) + self.logger.info( + f"Wikipedia library language set to '{wiki_lang}'. Rate limiting enabled (min_wait=0.1)." + ) except Exception as e: - # Logge Fehler bei der Konfiguration der wikipedia Bibliothek self.logger.warning(f"Fehler beim Setzen der Wikipedia-Sprache oder Rate Limiting: {e}") - - # --- Interne Helfermethoden --- - # Diese Methoden sind nur fuer die interne Nutzung der WikipediaScraper Klasse gedacht. - def _get_full_domain(self, website): """Extrahiert die normalisierte Domain (ohne www, ohne Pfad) aus einer URL.""" - # Diese Funktion nutzt die globale simple_normalize_url, was besser ist als eine duplizierte Implementierung. - return simple_normalize_url(website) # Nutzt die globale Funktion (Block 4) - + return simple_normalize_url(website) def _generate_search_terms(self, company_name, website): """ Generiert eine Liste von Suchbegriffen fuer die Wikipedia-Suche, inklusive normalisiertem Namen, Kurzformen und Domain. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - if not company_name: return [] # Gebe leere Liste zurueck, wenn kein Firmenname da ist - terms = set() # Nutzt ein Set, um Duplikate automatisch zu behandeln + if not company_name: + return [] + terms = set() - # Fuegen Sie den originalen Namen hinzu original_name_cleaned = str(company_name).strip() if original_name_cleaned: terms.add(original_name_cleaned) - # Fuegen Sie den normalisierten Namen und Teile davon hinzu (nutzt globale Funktion) - normalized_name = normalize_company_name(company_name) # Nutzt die globale Funktion (Block 4) + normalized_name = normalize_company_name(company_name) if normalized_name: - terms.add(normalized_name) - name_parts = normalized_name.split() - if len(name_parts) > 0: terms.add(name_parts[0]) # Erstes Wort - if len(name_parts) > 1: terms.add(" ".join(name_parts[:2])) # Erste zwei Woerter + terms.add(normalized_name) + name_parts = normalized_name.split() + if len(name_parts) > 0: + terms.add(name_parts[0]) + if len(name_parts) > 1: + terms.add(" ".join(name_parts[:2])) - # Fuegen Sie die Domain hinzu (nutzt interne Methode, die simple_normalize_url nutzt) full_domain = self._get_full_domain(website) - # Wenn die Domain gueltig ist (nicht "k.A.") if full_domain != "k.A.": - terms.add(full_domain) + terms.add(full_domain) - - # Entferne leere Strings aus dem Set und konvertiere zu einer Liste. - # Limitiere die Anzahl der Begriffe auf die konfigurierte Anzahl der Suchergebnisse. - final_terms = [term for term in list(terms) if term][:getattr(Config, 'WIKIPEDIA_SEARCH_RESULTS', 5)] # Limitiere auf Anzahl der Suchergebnisse (Config, Block 1) - # Logge die generierten Suchbegriffe auf Debug-Level - self.logger.debug(f"Generierte Suchbegriffe fuer '{company_name[:100]}...': {final_terms}") # Gekuerzt loggen + final_terms = [term for term in list(terms) if term][:getattr(Config, 'WIKIPEDIA_SEARCH_RESULTS', 5)] + self.logger.debug(f"Generierte Suchbegriffe fuer '{company_name[:100]}...': {final_terms}") return final_terms - - @retry_on_failure # Wende den Decorator an, da es externe Calls macht (Requests) + @retry_on_failure def _get_page_soup(self, url): """ Holt HTML von einer URL (requests) und gibt ein BeautifulSoup-Objekt zurueck. - Dies wird fuer die manuelle Extraktion von Infoboxen und anderen Details benoetigt, - die nicht direkt ueber die wikipedia Bibliothek verfuegbar sind. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefen Sie auf ungueltige oder leere URLs if not url or not isinstance(url, str) or not url.lower().startswith(("http://", "https://")): - self.logger.warning(f"_get_page_soup: Ungueltige URL '{url[:100]}...'.") # Gekuerzt loggen - return None # Gebe None zurueck bei ungueltigen Eingaben - - + self.logger.warning(f"_get_page_soup: Ungueltige URL '{url[:100]}...'.") + return None try: - # Logge den Versuch, die URL abzurufen - self.logger.debug(f"_get_page_soup: Rufe URL ab: {url[:100]}...") # Gekuerzt loggen - # Führen Sie die GET-Anfrage mit der Instanz Session durch (behaelt Cookies etc.). - # Timeout sollte aus Config kommen. - # Die raise_for_status() und die RequestsExceptions werden vom retry_on_failure Decorator behandelt. + self.logger.debug(f"_get_page_soup: Rufe URL ab: {url[:100]}...") response = self.session.get(url, timeout=getattr(Config, 'REQUEST_TIMEOUT', 15)) - response.raise_for_status() # Wirft HTTPError fuer 4xx/5xx Antworten. - - # Versuchen Sie, das Encoding aus dem Header oder dem Content zu erraten + response.raise_for_status() response.encoding = response.apparent_encoding - - # Parsen Sie den HTML-Inhalt mit BeautifulSoup. - # Nutzt den konfigurierten Parser aus Config oder einen Fallback. soup = BeautifulSoup(response.text, getattr(Config, 'HTML_PARSER', 'html.parser')) + self.logger.debug(f"_get_page_soup: Parsen von {url[:100]}... erfolgreich.") + return soup + except Exception as e: + self.logger.error(f"_get_page_soup: Fehler beim Abrufen oder Parsen von HTML von {url[:100]}...: {type(e).__name__} - {e}") + raise e - # Logge den Erfolg des Parsens - self.logger.debug(f"_get_page_soup: Parsen von {url[:100]}... erfolgreich.") # Gekuerzt loggen - return soup # Gebe das Soup-Objekt zurueck - - # Exceptions (wie RequestsErrors) werden vom retry_on_failure Decorator behandelt. - # Wenn eine Exception hier durchkommt, hat der Decorator aufgegeben. - except Exception as e: # Fangen Sie alle verbleibenden Exceptions ab - # Logge den Fehler auf Error-Level - self.logger.error(f"_get_page_soup: Fehler beim Abrufen oder Parsen von HTML von {url[:100]}...: {type(e).__name__} - {e}") # Gekuerzt loggen - # Die Exception wurde bereits vom retry_on_failure Decorator als finaler Fehler geloggt. - # Werfen Sie die Exception erneut, damit der Aufrufer dies behandeln kann (z.B. extract_company_data). - raise e # Leite die Exception weiter - - - # --- Überarbeitete Validierungsmethode --- - # Validiert, ob ein Wikipedia-Artikel zum Unternehmen passt. - # Nutzt interne Methoden und globale Helfer. def _validate_article(self, page, company_name, website): """ Validiert, ob ein Wikipedia-Artikel (represented by the Page object) - zum Unternehmen passt. Prueft Titelaehnlichkeit (gewichtet Anfangsworte), - Domain-Match in Links und passt Schwellenwerte dynamisch an. - - Args: - page (wikipedia.WikipediaPage): Das geladene Wikipedia Page Objekt. - company_name (str): Der Name des Unternehmens (CRM Name). - website (str): Die Website des Unternehmens (CRM Website). - - Returns: - bool: True, wenn der Artikel validiert wurde, sonst False. + zum Unternehmen passt. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - if not page or not company_name: return False # Grundlegende Pruefung - # page.title ist der Titel des Wikipedia-Artikels - self.logger.debug(f"Validiere Artikel '{page.title[:100]}...' (URL: {page.url[:100]}...) fuer Firma '{company_name[:100]}' (Website: {website[:100]})...") # Gekuerzt loggen + if not page or not company_name: + return False + self.logger.debug( + f"Validiere Artikel '{page.title[:100]}...' (URL: {page.url[:100]}...) " + f"fuer Firma '{company_name[:100]}' (Website: {website[:100]})..." + ) - - # Normalisiere Namen (nutzt globale Funktion Block 4) normalized_company = normalize_company_name(company_name) normalized_title = normalize_company_name(page.title) - - # Wenn Normalisierung fehlschlaegt oder zu leeren Strings fuehrt if not normalized_company or not normalized_title: - self.logger.warning("Validierung nicht moeglich, da Normalisierung eines Namens fehlschlug.") - return False + self.logger.warning("Validierung nicht moeglich, da Normalisierung eines Namens fehlschlug.") + return False - # Basisschwelle fuer Aehnlichkeit aus Config (Block 1) standard_threshold = getattr(Config, 'SIMILARITY_THRESHOLD', 0.65) - - # 1. Titelaehnlichkeit (Gesamt) - # Berechne Aehnlichkeit zwischen den normalisierten Namen (nutzt globale Helfer Block 4) similarity = fuzzy_similarity(normalized_title, normalized_company) - self.logger.debug(f" -> Gesamt-Aehnlichkeit (normalized): {similarity:.2f} ('{normalized_title[:50]}...' vs '{normalized_company[:50]}...')") # Gekuerzt loggen + self.logger.debug(f" -> Gesamt-Aehnlichkeit (normalized): {similarity:.2f}") - # 2. Aehnlichkeit der ersten Worte (Normalisiert) company_tokens = normalized_company.split() title_tokens = normalized_title.split() first_word_match = False first_two_words_match = False + if company_tokens and title_tokens and company_tokens[0] == title_tokens[0]: + first_word_match = True + if len(company_tokens) > 1 and len(title_tokens) > 1 and company_tokens[1] == title_tokens[1]: + first_two_words_match = True - if len(company_tokens) > 0 and len(title_tokens) > 0: - # Pruefe das erste Wort (case-sensitive nach Normalisierung) - if company_tokens[0] == title_tokens[0]: - first_word_match = True - # self.logger.debug(" -> Erstes normalisiertes Wort stimmt ueberein.") # Zu viel Laerm im Debug - # Wenn das erste Wort uebereinstimmt, pruefe das zweite Wort - if len(company_tokens) > 1 and len(title_tokens) > 1: - # Pruefe das zweite Wort (case-sensitive nach Normalisierung) - if company_tokens[1] == title_tokens[1]: - first_two_words_match = True - # self.logger.debug(" -> Erste zwei normalisierte Worte stimmen ueberein.") # Zu viel Laerm im Debug - - # 3. Link-Pruefung (Domain-Match in externen Links des Artikels) domain_found = False - # Extrahieren Sie die normalisierte Domain aus der Website-URL (nutzt interne Methode, die simple_normalize_url nutzt) full_domain = self._get_full_domain(website) - # Nur weitermachen, wenn eine gueltige Domain extrahiert wurde if full_domain != "k.A.": - self.logger.debug(f" -> Suche nach Domain '{full_domain}' in externen Links des Artikels...") - try: - # Direkte Abfrage des HTML ueber page.html() kann schneller sein als erneuter Requests Call - # page.html() kann Exceptions werfen. - article_html = page.html() - if article_html: - # Parsen Sie den HTML-Inhalt des Artikels - soup = BeautifulSoup(article_html, getattr(Config, 'HTML_PARSER', 'html.parser')) - # Suche nach externen Links (href, die mit http starten) - external_links = soup.select('a[href^="http"]') # Suche nach allen A-Tags mit href, der mit "http" beginnt + self.logger.debug(f" -> Suche nach Domain '{full_domain}' in externen Links des Artikels...") + try: + article_html = page.html() + if article_html: + soup = BeautifulSoup(article_html, getattr(Config, 'HTML_PARSER', 'html.parser')) + external_links = soup.select('a[href^="http"]') + relevant_links = [] + for link in external_links: + href = link.get('href', '') + if href and isinstance(href, str) and full_domain in simple_normalize_url(href): + if not any(ex in href.lower() for ex in ['wikipedia.org', 'wikimedia.org', 'wikidata.org', 'archive.org', 'webcitation.org']): + relevant_links.append(link) + if relevant_links: + domain_found = True + except Exception as e_link_check: + self.logger.error( + f"Fehler waehrend der Domain-Link-Pruefung fuer '{page.title[:100]}...': " + f"{type(e_link_check).__name__} - {e_link_check}" + ) - # Filtere Links: Muss die gesuchte Domain enthalten UND darf kein Wikipedia-eigener Link sein. - relevant_links = [] - for link in external_links: - href = link.get('href', '') # Hole das href-Attribut - # Pruefe, ob href gueltig ist und die Domain im normalisierten Link enthalten ist - if href and isinstance(href, str) and full_domain in simple_normalize_url(href): - # Pruefe, ob der Link NICHT auf eine Wikipedia-eigene Seite verweist - if not any(exclude in href.lower() for exclude in ['wikipedia.org', 'wikimedia.org', 'wikidata.org', 'archive.org', 'webcitation.org']): - relevant_links.append(link) # Fuege den relevanten Link zur Liste hinzu - - # Wenn relevante Links gefunden wurden - if relevant_links: - # Optional: Pruefe, ob der Link in der Infobox ist oder typischen Text hat (Komplexitaet vs Nutzen) - # Fuer eine einfachere Implementierung reicht es, wenn der Link existiert. - domain_found = True - # logger.debug(f" -> Domain '{full_domain}' in {len(relevant_links)} externen Links gefunden.") # Zu viel Laerm im Debug - else: - # logger.debug(f" -> Domain '{full_domain}' nicht in externen Links gefunden.") # Zu viel Laerm im Debug - pass # domain_found bleibt False - - else: - self.logger.warning(" -> Konnte HTML fuer Link-Pruefung nicht abrufen (page.html() leer).") - - except Exception as e_link_check: - # Logge Fehler waehrend der Domain-Link-Pruefung, aber fahre fort - self.logger.error(f"Fehler waehrend der Domain-Link-Pruefung fuer '{page.title[:100]}...': {type(e_link_check).__name__} - {e_link_check}") # Gekuerzt loggen - # Fehler beim Link-Check sollte die Validierung nicht blockieren, nur beeinflussen. - pass # domain_found bleibt False oder sein aktueller Wert. - - else: - self.logger.debug(" -> Keine Website-Domain fuer Link-Pruefung vorhanden oder ungueltig ('k.A.').") - - - # 4. Dynamische Schwellenwert-Entscheidung (Bewertung des Artikels) - is_valid = False # Initialisierung des Validierungs-Flags - reason = "Keine Validierungsregel traf zu" # Default Grund fuer das Ergebnis - - # Pruefe die Validierungsbedingungen in absteigender Reihenfolge ihrer Stärke / Relevanz - # Die erste Bedingung, die True ist, bestimmt das Ergebnis. + # Dynamische Schwellenwert-Entscheidung + is_valid = False + reason = "Keine Validierungsregel traf zu" if similarity >= standard_threshold: is_valid = True reason = f"Gesamt-Aehnlichkeit ({similarity:.2f}) >= Standard-Schwelle ({standard_threshold:.2f})" - elif domain_found and first_two_words_match: # Staerkste Kombination von Indikatoren - is_valid = True - reason = f"Domain gefunden UND erste 2 normalisierte Worte stimmen ueberein (Sim={similarity:.2f})" - elif domain_found and first_word_match and similarity >= 0.40: # Domain + Erstes Wort + Moderate Aehnlichkeit - is_valid = True - reason = f"Domain gefunden UND erstes normalisiertes Wort stimmt ueberein UND Aehnlichkeit >= 0.40 (Sim={similarity:.2f})" - elif first_two_words_match and similarity >= 0.45: # Erste zwei Worte + Moderate Aehnlichkeit (auch ohne Domain) - is_valid = True - reason = f"Erste zwei normalisierte Worte stimmen ueberein UND Aehnlichkeit >= 0.45 (Sim={similarity:.2f})" - elif domain_found and similarity >= 0.50: # Nur Domain + Etwas hoehere Aehnlichkeit - is_valid = True - reason = f"Domain gefunden UND Aehnlichkeit >= 0.50 (Sim={similarity:.2f})" - elif first_word_match and similarity >= 0.55: # Nur Erstes Wort + Etwas hoehere Aehnlichkeit - is_valid = True - reason = f"Erstes normalisiertes Wort stimmt ueberein UND Aehnlichkeit >= 0.55 (Sim={similarity:.2f})" - # Weitere optionale, weniger strenge Schwellen koennten hier hinzugefuegt werden, wenn gewuenscht. - # z.B. elif domain_found and similarity >= 0.30: + elif domain_found and first_two_words_match: + is_valid = True + reason = f"Domain gefunden UND erste 2 normalisierte Worte stimmen ueberein (Sim={similarity:.2f})" + elif domain_found and first_word_match and similarity >= 0.40: + is_valid = True + reason = f"Domain gefunden UND erstes normalisiertes Wort stimmt ueberein UND Aehnlichkeit >= 0.40 (Sim={similarity:.2f})" + elif first_two_words_match and similarity >= 0.45: + is_valid = True + reason = f"Erste zwei normalisierte Worte stimmen ueberein UND Aehnlichkeit >= 0.45 (Sim={similarity:.2f})" + elif domain_found and similarity >= 0.50: + is_valid = True + reason = f"Domain gefunden UND Aehnlichkeit >= 0.50 (Sim={similarity:.2f})" + elif first_word_match and similarity >= 0.55: + is_valid = True + reason = f"Erstes normalisiertes Wort stimmt ueberein UND Aehnlichkeit >= 0.55 (Sim={similarity:.2f})" - - # Logge das Ergebnis der Validierung (INFO wenn validiert, DEBUG sonst) log_level = logging.INFO if is_valid else logging.DEBUG - self.logger.log(log_level, f" => Artikel '{page.title[:100]}...' {'VALIDIERT' if is_valid else 'NICHT validiert'} (Grund: {reason}. Details: Sim={similarity:.2f}, Domain? {domain_found}, 1stWord? {first_word_match}, 2ndWord? {first_two_words_match})") # Gekuerzt loggen - - return is_valid # Gebe das Ergebnis der Validierung zurueck - - - # --- Extraktionsmethoden --- - # Methoden zum Extrahieren spezifischer Daten aus einem Wikipedia Soup-Objekt. + self.logger.log( + log_level, + f" => Artikel '{page.title[:100]}...' " + f"{'VALIDIERT' if is_valid else 'NICHT validiert'} " + f"(Grund: {reason}. Details: Sim={similarity:.2f}, Domain? {domain_found}, " + f"1stWord? {first_word_match}, 2ndWord? {first_two_words_match})" + ) + return is_valid def _extract_first_paragraph_from_soup(self, soup): """ Extrahiert den ersten aussagekraeftigen Absatz aus dem Soup-Objekt eines Wikipedia-Artikels. Entfernt Referenzen und unerwuenschte Elemente. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - if not soup: return "k.A." # Gebe "k.A." zurueck, wenn kein Soup-Objekt da ist - paragraph_text = "k.A." # Initialisiere mit "k.A." + if not soup: + return "k.A." + paragraph_text = "k.A." try: - # Finden Sie den Hauptinhalt-Div (mw-parser-output ist Standard auf moderner Wikipedia) content_div = soup.find('div', class_='mw-parser-output') - # Wenn der Haupt-Div gefunden wurde, suchen Sie darin, sonst im gesamten Soup search_area = content_div if content_div else soup - # Suchen Sie die ersten p-Tags direkt unter diesem Div (recursive=False) oder im gesamten Suchbereich. - # Oft ist der erste relevante Absatz ein direktes Kind des Haupt-Divs. paragraphs = search_area.find_all('p', recursive=False) - # Fallback, falls keine direkten p-Tags gefunden werden, suche rekursiv im gesamten Suchbereich. - if not paragraphs: paragraphs = search_area.find_all('p') + if not paragraphs: + paragraphs = search_area.find_all('p') - - # Gehe durch die gefundenen Absaetze, um den ersten geeigneten zu finden - # logger.debug(f"Suche ersten Absatz in {len(paragraphs)} gefundenen

-Tags...") # Zu viel Laerm im Debug for p in paragraphs: - # Entferne Stoerende Elemente wie Referenzen ([1]), versteckte Spans oder Koordinaten innerhalb des p-Tags, BEVOR der Text extrahiert wird. - for sup in p.find_all('sup', class_='reference'): sup.decompose() # Referenz-Tags - for span in p.find_all('span', style=lambda value: value and 'display:none' in value): span.decompose() # Versteckte Spans (z.B. fuer Sortierschluessel) - for span in p.find_all('span', id='coordinates'): span.decompose() # Koordinaten-Span + for sup in p.find_all('sup', class_='reference'): + sup.decompose() + for span in p.find_all('span', style=lambda v: v and 'display:none' in v): + span.decompose() + for span in p.find_all('span', id='coordinates'): + span.decompose() - - # Extrahiere und bereinige den Text des Absatzes (nutzt globale Funktion clean_text Block 4) text = clean_text(p.get_text(separator=' ', strip=True)) + if text != "k.A." and len(text) > 50: + if not re.match( + r'^(Datei:|Abbildung:|Siehe auch:|Einzelnachweise|Siehe auch|Literatur)', + text, + re.IGNORECASE + ): + paragraph_text = text[:1500] + break - # Pruefe, ob der Text nach der Bereinigung gueltig und lang genug ist. - # Ein sehr kurzer Text ist oft nur eine Bildunterschrift oder aehnliches. - if text != "k.A." and len(text) > 50: # Mindestlaenge 50 Zeichen (kann angepasst werden) - # Pruefe auf gaengige unerwuenschte Anfange (z.B. nach Infoboxen oder Listen) - # Vermeide Absaetze, die mit "Datei:", "Abbildung:", "Siehe auch:", "Einzelnachweise" etc. beginnen. - if not re.match(r'^(Datei:|Abbildung:|Siehe auch:|Einzelnachweise|Siehe auch|Literatur|Einzelnachweise)', text, re.IGNORECASE): - # Wenn der Absatz relevant aussieht, nimm ihn und begrenze die Laenge. - paragraph_text = text[:1500] # Begrenze die Laenge des extrahierten Absatzes (z.B. auf 1500 Zeichen) - # logger.debug(f" -> Ersten gueltigen Absatz gefunden: {paragraph_text[:100]}...") # Logge den Anfang des Absatzes (gekuerzt) - break # Hoere beim ersten guten Absatz auf - - - # Wenn die Schleife durchlaeuft und kein passender Absatz gefunden wurde if paragraph_text == "k.A.": - self.logger.debug("Kein passender erster Absatz gefunden nach Pruefung der

-Tags.") - + self.logger.debug("Kein passender erster Absatz gefunden nach Pruefung der

-Tags.") except Exception as e: - # Fange unerwartete Fehler waehrend der Extraktion ab self.logger.error(f"Fehler beim Extrahieren des ersten Absatzes: {type(e).__name__} - {e}") - # Logge den Traceback self.logger.debug(traceback.format_exc()) - # paragraph_text bleibt "k.A." oder sein Initialwert - - return paragraph_text # Gebe den extrahierten Text oder "k.A." zurueck - + return paragraph_text def extract_categories(self, soup): """ Extrahiert Wikipedia-Kategorien aus dem Soup-Objekt. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - if not soup: return "k.A." # Gebe "k.A." zurueck, wenn kein Soup-Objekt da ist - cats_filtered = [] # Liste zum Sammeln der bereinigten Kategorien + if not soup: + return "k.A." + cats_filtered = [] try: - # Kategorien sind normalerweise in einem div mit id="mw-normal-catlinks" cat_div = soup.find('div', id="mw-normal-catlinks") - # Wenn der Kategorien-Div gefunden wurde if cat_div: - # Die Kategorien sind innerhalb eines ul-Tags unter diesem div ul = cat_div.find('ul') - # Wenn das ul-Tag gefunden wurde if ul: - # Jede Kategorie ist ein li-Element innerhalb des ul. - # Extrahieren Sie den Text und bereinigen Sie ihn (clean_text nutzt globale Funktion Block 4). cats = [clean_text(li.get_text()) for li in ul.find_all('li')] - # Filtere leere oder unerwuenschte Eintraege (wie "Kategorien:") aus der Liste - cats_filtered = [c for c in cats if c and isinstance(c, str) and c.strip() and "kategorien:" not in c.lower()] - # Logge die gefundenen Kategorien auf Debug-Level + cats_filtered = [ + c for c in cats + if c and isinstance(c, str) and c.strip() and "kategorien:" not in c.lower() + ] self.logger.debug(f"Kategorien gefunden: {cats_filtered}") else: - # Wenn kein ul-Tag im Kategorien-Div gefunden wurde - self.logger.debug("Kein 'ul' Tag in 'mw-normal-catlinks' gefunden.") + self.logger.debug("Kein 'ul' Tag in 'mw-normal-catlinks' gefunden.") else: - # Wenn der Kategorien-Div nicht gefunden wurde - self.logger.debug("Kein 'div#mw-normal-catlinks' gefunden.") - + self.logger.debug("Kein 'div#mw-normal-catlinks' gefunden.") except Exception as e: - # Fange unerwartete Fehler waehrend der Extraktion ab - self.logger.error(f"Fehler beim Extrahieren der Kategorien: {type(e).__name__} - {e}") - # Logge den Traceback - self.logger.debug(traceback.format_exc()) - - # Verbinde die bereinigten Kategorien mit Komma und Leerzeichen, oder gebe "k.A." zurueck, wenn keine gefunden wurden. + self.logger.error(f"Fehler beim Extrahieren der Kategorien: {type(e).__name__} - {e}") + self.logger.debug(traceback.format_exc()) return ", ".join(cats_filtered) if cats_filtered else "k.A." - def _extract_infobox_value(self, soup, target): """ Extrahiert gezielt Branche, Umsatz oder Mitarbeiter aus der Infobox eines Wikipedia-Artikels Soup-Objekts. - Beruecksichtigt Header in oder fett formatierten . - Nutzt interne Keywords-Map und globale Helfer (clean_text, extract_numeric_value, logger, re). """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist self.logger.debug(f"--- Entering _extract_infobox_value for target '{target}' ---") - # Pruefen Sie, ob ein Soup-Objekt vorhanden ist und das Ziel in der Keywords-Map existiert if not soup or target not in self.keywords_map: - self.logger.debug(f"_extract_infobox_value: Ungueltiger Input (Soup: {soup is not None}, Target: {target})") - return "k.A." # Gebe "k.A." zurueck bei ungueltigen Eingaben - - - keywords = self.keywords_map[target] # Holen Sie die Keywords fuer das Ziel + self.logger.debug( + f"_extract_infobox_value: Ungueltiger Input (Soup: {soup is not None}, Target: {target})" + ) + return "k.A." + keywords = self.keywords_map[target] self.logger.debug(f"_extract_infobox_value: Suche nach '{target}' mit Keywords: {keywords}") - # Finden Sie die Infobox (verschiedene Klassen sind moeglich) - # Nutzt select_one, um das erste passende Element zu finden - infobox = soup.select_one('table[class*="infobox"]') # Suche nach table mit class, die "infobox" enthaelt - - # Wenn keine Infobox gefunden wurde + infobox = soup.select_one('table[class*="infobox"]') if not infobox: self.logger.debug(" -> KEINE Infobox via select_one 'table[class*=\"infobox\"]' gefunden.") - return "k.A." # Gebe "k.A." zurueck, wenn keine Infobox gefunden wurde - - # Logge, dass eine Infobox gefunden wurde - self.logger.debug(f" -> Infobox gefunden.") - value_found = "k.A." # Initialisiere das Ergebnis mit "k.A." + return "k.A." + self.logger.debug(" -> Infobox gefunden.") + value_found = "k.A." try: - # Iteriere durch die Zeilen (tr) der Infobox rows = infobox.find_all('tr') - # logger.debug(f" -> Analysiere {len(rows)} Zeilen in der Infobox.") # Zu viel Laerm im Debug - - # Durchsuche jede Zeile nach einem Header-Zelle und einer Wert-Zelle for idx, row in enumerate(rows): - # logger.debug(f" --- Pruefe Roh-HTML Zeile {idx}: {str(row)[:150]}...") # Zu viel Laerm im Debug - - # Suche nach TH und TD Elementen direkt unter TR (recursive=False) cells = row.find_all(['th', 'td'], recursive=False) - header_text = None # Initialisiere Header-Text und Wert-Zelle + header_text = None value_cell = None - # Gaengigste Struktur: TH (Header) gefolgt von TD (Wert) if len(cells) >= 2 and cells[0].name == 'th' and cells[1].name == 'td': - header_text = cells[0].get_text(strip=True) # Extrahiere Text aus TH - value_cell = cells[1] # TD ist die Wert-Zelle - # logger.debug(f" -> Zeile {idx}: Struktur TH + TD erkannt.") # Zu viel Laerm im Debug - - # Alternative Struktur: TD (Header-aehnlich, z.B. fett) gefolgt von TD (Wert) - # Hier ist Vorsicht geboten, um nicht regulaere Datenzellen zu erfassen. + header_text = cells[0].get_text(strip=True) + value_cell = cells[1] elif len(cells) >= 2 and cells[0].name == 'td' and cells[1].name == 'td': - first_cell_is_header_like = False - # Pruefe auf Style-Attribut mit font-weight bold - style = cells[0].get('style', '').lower() - if 'font-weight' in style and ('bold' in style or '700' in style): - first_cell_is_header_like = True - # Pruefe auf fettgedruckten Inhalt ( oder ) - elif cells[0].find(['b', 'strong'], recursive=False): - first_cell_is_header_like = True + style = cells[0].get('style', '').lower() + first_cell_is_header_like = False + if 'font-weight' in style and ('bold' in style or '700' in style): + first_cell_is_header_like = True + elif cells[0].find(['b', 'strong'], recursive=False): + first_cell_is_header_like = True + if first_cell_is_header_like: + header_text = cells[0].get_text(strip=True) + value_cell = cells[1] - if first_cell_is_header_like: - header_text = cells[0].get_text(strip=True) # Extrahiere Text aus dem TD, das wie ein Header aussieht - value_cell = cells[1] # Das zweite TD ist die Wert-Zelle - # logger.debug(f" -> Zeile {idx}: Struktur TD(Header-like) + TD erkannt.") # Zu viel Laerm im Debug - # else: - # logger.debug(f" -> Zeile {idx}: Struktur TD + TD, aber erstes TD nicht als Header erkannt.") # Zu viel Laerm im Debug - - # Wenn eine passende Struktur (Header + Wert-Zelle) gefunden wurde if header_text is not None and value_cell is not None: - # logger.debug(f" -> Verarbeite Zeile {idx} mit Header='{header_text}'") # Zu viel Laerm im Debug header_text_lower = header_text.lower() - matched_keyword = None # Initialisiere das gefundene Keyword - - # Pruefe, ob ein gesuchtes Keyword im Header-Text (kleingeschrieben) vorkommt + matched_keyword = None for kw in keywords: if kw in header_text_lower: - matched_keyword = kw # Speichere das gefundene Keyword - break # Hoere auf, sobald ein Keyword gefunden wurde - - # Wenn ein Keyword gefunden wurde, extrahiere den Wert aus der Wert-Zelle + matched_keyword = kw + break if matched_keyword: - # logger.debug(f" --> Keyword '{matched_keyword}' gefunden in Header '{header_text}'!") # Logge das gefundene Keyword - - # Entferne stoerende Elemente wie Referenz-Tags ([1]) oder versteckte Spans innerhalb der Value-Zelle, BEVOR der Text extrahiert wird. for sup in value_cell.find_all(['sup', 'span']): - # Pruefe, ob es ein Referenz-Tag ODER ein Span mit display:none Style ist. - if (sup.name == 'sup' and sup.has_attr('class') and 'reference' in sup['class']) or \ - (sup.name == 'span' and sup.get('style') and 'display:none' in sup['style']): - # logger.debug(f" -> Entferne stoerendes Element: {sup.get_text(strip=True)[:50]}...") # Logge Entfernung (gekuerzt) - sup.decompose() # Entferne das Element aus dem Baum + if ( + (sup.name == 'sup' and sup.has_attr('class') and 'reference' in sup['class']) + or (sup.name == 'span' and sup.get('style') and 'display:none' in sup['style']) + ): + sup.decompose() - # Extrahiere den Rohtext aus der bereinigten Value-Zelle - # Verwenden Sie Leerzeichen als Trenner zwischen Elementen im Text raw_value_text = value_cell.get_text(separator=' ', strip=True) - # logger.debug(f" -> Roher TD/Value-Text nach Decompose: '{raw_value_text[:100]}...'") # Logge Rohtext (gekuerzt) - - # Bereinige und konvertiere den Wert basierend auf dem Zieltyp (Branche, Umsatz, Mitarbeiter) if target == 'branche': - # Branche: Bereinigen, Klammern entfernen, nur erste Zeile nehmen - clean_val = clean_text(raw_value_text) # Nutzt globale Funktion clean_text (Block 4) - clean_val = re.sub(r'\s*\([^)]*\)', '', clean_val).strip() # Entferne Text in runden Klammern (z.B. "(Stand 2022)") - clean_val = clean_val.split('\n')[0].strip() # Nimm nur die erste Zeile (falls es mehrere Zeilen gibt) - # Setze den gefundenen Wert oder "k.A." wenn er leer ist + clean_val = clean_text(raw_value_text) + clean_val = re.sub(r'\s*\([^)]*\)', '', clean_val).strip() + clean_val = clean_val.split('\n')[0].strip() value_found = clean_val if clean_val else "k.A." - self.logger.info(f" --> Branche extrahiert: '{value_found}'") # Logge das Ergebnis - + self.logger.info(f" --> Branche extrahiert: '{value_found}'") elif target == 'umsatz': - # Umsatz: Numerische Extraktion (nutzt globale Funktion extract_numeric_value Block 5) - # extract_numeric_value gibt einen String zurueck ("k.A." oder Zahl) numeric_val_str = extract_numeric_value(raw_value_text, is_umsatz=True) - value_found = numeric_val_str # Setze den gefundenen Wert (String) - self.logger.info(f" --> Umsatz extrahiert (aus '{raw_value_text[:50]}...'): '{value_found}'") # Logge Ergebnis (gekuerzt) - + value_found = numeric_val_str + self.logger.info( + f" --> Umsatz extrahiert (aus '{raw_value_text[:50]}...'): '{value_found}'" + ) elif target == 'mitarbeiter': - # Mitarbeiter: Numerische Extraktion (nutzt globale Funktion extract_numeric_value Block 5) - # extract_numeric_value gibt einen String zurueck ("k.A." oder Zahl) numeric_val_str = extract_numeric_value(raw_value_text, is_umsatz=False) - value_found = numeric_val_str # Setze den gefundenen Wert (String) - self.logger.info(f" --> Mitarbeiter extrahiert (aus '{raw_value_text[:50]}...'): '{value_found}'") # Logge Ergebnis (gekuerzt) - - # Da wir den Wert fuer das gesuchte Ziel gefunden haben, koennen wir die Schleife ueber die Zeilen abbrechen + value_found = numeric_val_str + self.logger.info( + f" --> Mitarbeiter extrahiert (aus '{raw_value_text[:50]}...'): '{value_found}'" + ) break - # Wenn die Schleife durchlaeuft und kein passendes Keyword gefunden wurde, bleibt value_found "k.A." if value_found != "k.A.": - self.logger.debug(f" -> Finaler Wert fuer '{target}' gefunden: '{value_found}'") + self.logger.debug(f" -> Finaler Wert fuer '{target}' gefunden: '{value_found}'") else: - self.logger.debug(f" -> Kein passender Eintrag fuer '{target}' in der gesamten Infobox gefunden.") - - + self.logger.debug( + f" -> Kein passender Eintrag fuer '{target}' in der gesamten Infobox gefunden." + ) except Exception as e: - # Fange jeden unerwarteten Fehler ab, der waehrend der Infobox-Verarbeitung auftritt - self.logger.exception(f"Fehler beim Durchlaufen der Infobox-Zeilen fuer '{target}': {e}") # Logge den Fehler und Traceback - return "k.A. (Fehler Extraktion)" # Gebe einen Fehlerwert zurueck bei Fehler + self.logger.exception( + f"Fehler beim Durchlaufen der Infobox-Zeilen fuer '{target}': {e}" + ) + return "k.A. (Fehler Extraktion)" - return value_found # Gebe den extrahierten Wert oder "k.A."/"Fehler" zurueck + return value_found - - # --- Hauptmethoden der Klasse --- - # Methoden zum Suchen und Extrahieren von Wikipedia-Daten. - - @retry_on_failure # Wende den Decorator an, da es externe Calls macht (wikipedia Bibliothek nutzt Requests/API) + @retry_on_failure def search_company_article(self, company_name, website=None): """ Sucht einen passenden Wikipedia-Artikel fuer das Unternehmen und gibt das wikipedia.WikipediaPage Objekt zurueck, wenn ein relevanter und validierter Artikel gefunden wird. Behandelt explizit Begriffsklaerungsseiten. - - Args: - company_name (str): Der Name des Unternehmens (CRM Name). - website (str, optional): Die Website des Unternehmens (CRM Website, fuer Kontext). Defaults to None. - - Returns: - wikipedia.WikipediaPage: Das validierte Page Objekt oder None. - Wirft Exception bei API/Netzwerk-Fehlern nach Retries. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Pruefe, ob ein Firmenname da ist if not company_name or str(company_name).strip() == "": self.logger.warning("Wikipedia search skipped: No company name provided.") - # Werfen Sie einen ValueError raise ValueError("Kein Firmenname fuer Wikipedia Suche angegeben.") - - # Generiere Suchbegriffe (nutzt interne Methode, die globale Helfer nutzt) search_terms = self._generate_search_terms(company_name, website) if not search_terms: - self.logger.warning(f"Keine Suchbegriffe fuer '{company_name[:100]}...' generiert.") # Gekuerzt loggen - return None # Gebe None zurueck, wenn keine Suchbegriffe generiert werden konnten + self.logger.warning(f"Keine Suchbegriffe fuer '{company_name[:100]}...' generiert.") + return None - # Logge den Start der Suche - self.logger.info(f"Starte Wikipedia-Suche fuer '{company_name[:100]}...' (Website: {website[:100]}...) mit Begriffen: {search_terms}") # Gekuerzt loggen - - # Menge der bereits geprueften Titel, um Redundanzen zu vermeiden und Endlosschleifen bei Weiterleitungen/Begriffs-k. zu verhindern. + self.logger.info( + f"Starte Wikipedia-Suche fuer '{company_name[:100]}...' " + f"(Website: {website[:100]}...) mit Begriffen: {search_terms}" + ) processed_titles = set() - # --- Innere Helferfunktion zum Pruefen eines einzelnen Titels --- - # Diese Funktion ist rekursiv (kann sich selbst aufrufen, z.B. bei Begriffsklaerungen). def check_page(title_to_check): - """Laedt einen potenziellen Wikipedia-Artikel und validiert ihn.""" - # Pruefen, ob der Titel bereits verarbeitet wurde (innerhalb dieses search_company_article Aufrufs) if title_to_check in processed_titles: - # self.logger.debug(f" -> Titel '{title_to_check[:100]}...' bereits geprueft, ueberspringe.") # Zu viel Laerm im Debug (gekuerzt loggen) - return None # Titel wurde bereits geprüft, ueberspringe - - # Fuege den Titel zur Menge der verarbeiteten hinzu, BEVOR er geladen wird. + return None processed_titles.add(title_to_check) - self.logger.debug(f" -> Pruefe potenziellen Artikel: '{title_to_check[:100]}...'") # Gekuerzt loggen - + self.logger.debug(f" -> Pruefe potenziellen Artikel: '{title_to_check[:100]}...'") try: - # Lade die Seite ueber die wikipedia Bibliothek. - # auto_suggest=False deaktiviert automatische Titelkorrektur. - # preload=True laedt den Inhalt und die Infobox gleich mit (nuetzlich fuer Validierung/Extraktion). - # Dieser Aufruf kann verschiedene wikipedia.exceptions (PageError, DisambiguationError) oder RequestsExceptions werfen. page = wikipedia.page(title_to_check, auto_suggest=False, preload=True) - - # Pruefe, ob es sich um eine Begriffsklaerungsseite handelt (wird von wikipedia.page selbst als Exception geworfen) - # oder ob unsere Validierung fehlschlaegt - # Rufen Sie die interne Validierungsmethode auf if self._validate_article(page, company_name, website): - # Wenn der Artikel validiert wurde, geben Sie das Page-Objekt zurueck - self.logger.info(f" -> Titel '{page.title[:100]}...' erfolgreich validiert!") # Gekuerzt loggen - return page # Gebe das validierte Page-Objekt zurueck + self.logger.info(f" -> Titel '{page.title[:100]}...' erfolgreich validiert!") + return page else: - # Wenn der Artikel nicht validiert wurde - self.logger.debug(f" -> Titel '{title_to_check[:100]}...' nicht validiert.") # Gekuerzt loggen - return None # Gebe None zurueck, wenn nicht validiert - - # Spezifische Behandlung von wikipedia.exceptions + self.logger.debug(f" -> Titel '{title_to_check[:100]}...' nicht validiert.") + return None except wikipedia.exceptions.PageError: - # Titel existiert nicht auf Wikipedia (404 aehnlich) - self.logger.debug(f" -> Seite '{title_to_check[:100]}...' nicht gefunden (PageError).") # Gekuerzt loggen - return None # Gebe None zurueck, Seite nicht gefunden - + self.logger.debug(f" -> Seite '{title_to_check[:100]}...' nicht gefunden (PageError).") + return None except wikipedia.exceptions.DisambiguationError as e_inner: - # Titel fuehrt zu einer Begriffsklaerungsseite - self.logger.info(f" -> Begriffsklaerung '{title_to_check[:100]}...' gefunden. Pruefe Optionen: {str(e_inner.options)[:100]}...") # Logge Optionen (gekuerzt) - best_option_page = None # Initialisiere mit None - - # Gehe durch die Optionen der Begriffsklaerungsseite + self.logger.info( + f" -> Begriffsklaerung '{title_to_check[:100]}...' gefunden. " + f"Pruefe Optionen: {str(e_inner.options)[:100]}..." + ) for option in e_inner.options: - option_lower = option.lower() - # Filtere Optionen, die wahrscheinlich keine Unternehmensartikel sind (z.B. Personen, Orte, Listen). - # Fuegen Sie hier weitere Filter oder eine verbesserte Heuristik hinzu. - if any(exclude_word in option_lower for exclude_word in ["(person)", "(ort)", "(geographie)", "liste ", "liste)"]): - # logger.debug(f" -> Option uebersprungen (wahrscheinlich keine Firma): '{option[:100]}...'") # Zu viel Laerm im Debug (gekuerzt loggen) - continue # Ueberspringe diese Option - - # Checken Sie die Option rekursiv mit check_page. - # Dies wird die Option laden und validieren. - validated_option_page = check_page(option) - - # Wenn eine Option validiert wurde, nehmen Sie die erste als besten Treffer (oder implementieren Sie eine Ranking-Logik) - if validated_option_page: - self.logger.info(f" -> Option '{option[:100]}...' aus Begriffsklaerung erfolgreich validiert!") # Gekuerzt loggen - # Wir koennten hier aufhoeren (return validated_option_page) oder weiter nach dem BESTEN Treffer suchen. - # Fuer den Anfang nehmen wir den ersten validierten Treffer. - return validated_option_page # Gebe das validierte Page-Objekt zurueck - - # Wenn keine Option aus der Begriffsklaerung validiert wurde - self.logger.debug(f" -> Keine passende/validierte Unternehmens-Option in Begriffsklaerung '{title_to_check[:100]}...' gefunden.") # Gekuerzt loggen - return None # Gebe None zurueck, keine passende Option gefunden - - # Fangen Sie Netzwerkfehler oder Wikipedia-spezifische API-Fehler ab. - # Diese werden vom @retry_on_failure Decorator (auf search_company_article) behandelt. - # Wenn sie innerhalb von check_page auftreten, brechen sie NUR die Pruefung dieses EINEN Titels ab. - # Wir wollen, dass der Hauptaufruf (search_company_article) retried wird, nicht check_page. - # Also loggen wir hier den Fehler auf Warning und geben None zurueck, OHNE die Exception weiterzuwerfen. + option_lower = option.lower() + if any( + ex in option_lower + for ex in ["(person)", "(ort)", "(geographie)", "liste ", "liste)"] + ): + continue + validated_option_page = check_page(option) + if validated_option_page: + self.logger.info( + f" -> Option '{option[:100]}...' aus Begriffsklaerung erfolgreich validiert!" + ) + return validated_option_page + self.logger.debug( + f" -> Keine passende/validierte Unternehmens-Option in Begriffsklaerung '{title_to_check[:100]}...' gefunden." + ) + return None except (requests.exceptions.RequestException, wikipedia.exceptions.WikipediaException) as e_req: - self.logger.warning(f" -> Netzwerk/API-Fehler beim Laden/Validieren von '{title_to_check[:100]}...': {type(e_req).__name__} - {e_req}. Ueberspringe diesen Titel.") # Gekuerzt loggen - # Optional: Kleine Pause bei Netzwerkfehlern, um API nicht weiter zu reizen - # time.sleep(0.5) - return None # Gebe None zurueck, diesen Titel ueberspringen und naechsten versuchen - + self.logger.warning( + f" -> Netzwerk/API-Fehler beim Laden/Validieren von '{title_to_check[:100]}...': " + f"{type(e_req).__name__} - {e_req}. Ueberspringe diesen Titel." + ) + return None except Exception as e_page: - # Fangen Sie andere unerwartete Fehler bei der Seitenverarbeitung ab - self.logger.error(f" -> Unerwarteter Fehler bei Verarbeitung von Titel '{title_to_check[:100]}...': {type(e_page).__name__} - {e_page}") # Gekuerzt loggen - # Logge Traceback fuer unerwartete Fehler - self.logger.debug(traceback.format_exc()) - return None # Gebe None zurueck, diesen Titel ueberspringen + self.logger.error( + f" -> Unerwarteter Fehler bei Verarbeitung von Titel '{title_to_check[:100]}...': " + f"{type(e_page).__name__} - {e_page}" + ) + self.logger.debug(traceback.format_exc()) + return None - - # --- Haupt-Suchlogik (Iteriere durch Suchbegriffe und Ergebnisse) --- - self.logger.debug(f" -> Versuche direkten Match fuer '{company_name[:100]}...'...") # Gekuerzt loggen - # Versuche zuerst den exakten Firmennamen als Titel zu laden und zu validieren - # Rufe die innere Helferfunktion check_page auf + self.logger.debug(f" -> Versuche direkten Match fuer '{company_name[:100]}...'...") validated_page = check_page(company_name) if validated_page: - return validated_page # Direkter, validierter Treffer gefunden! + return validated_page - - self.logger.debug(f" -> Kein direkter Treffer/validiert. Starte Suche mit generierten Begriffen: {search_terms}") - # Wenn kein direkter Treffer, fuehre eine Suche mit den generierten Begriffen durch + self.logger.debug( + f" -> Kein direkter Treffer/validiert. Starte Suche mit generierten Begriffen: {search_terms}" + ) for term in search_terms: try: - self.logger.debug(f" -> Suche mit Begriff: '{term[:100]}...'...") # Gekuerzt loggen - # Fuehre die Suche ueber die wikipedia-Bibliothek durch - # wikipedia.search kann Exceptions werfen (z.B. PageError), die vom retry_on_failure im Decorator gefangen werden. - # results=getattr(Config, 'WIKIPEDIA_SEARCH_RESULTS', 5) holt die Anzahl aus Config (Block 1) + self.logger.debug(f" -> Suche mit Begriff: '{term[:100]}...'...") search_results = wikipedia.search(term, results=getattr(Config, 'WIKIPEDIA_SEARCH_RESULTS', 5)) - self.logger.debug(f" -> Suchergebnisse fuer '{term[:100]}...': {search_results}") # Gekuerzt loggen - - # Wenn keine Suchergebnisse fuer diesen Begriff gefunden wurden + self.logger.debug(f" -> Suchergebnisse fuer '{term[:100]}...': {search_results}") if not search_results: - self.logger.debug(f" -> Keine Suchergebnisse fuer '{term[:100]}...'.") # Gekuerzt loggen - continue # Springe zum naechsten Suchbegriff - - # Pruefe jeden Titel in den Suchergebnissen + self.logger.debug(f" -> Keine Suchergebnisse fuer '{term[:100]}...'.") + continue for title in search_results: - # Rufe die innere Helferfunktion check_page auf, um den Artikel zu laden und zu validieren - validated_page = check_page(title) - # Wenn ein validierter Artikel gefunden wurde - if validated_page: - return validated_page # Gebe den ersten validierten Artikel zurueck! - - # Kleine Pause zwischen dem Pruefen einzelner Suchergebnisse - # time.sleep(0.05) # Sehr kurz, optional - - # Fangen Sie Exceptions, die waehrend wikipedia.search auftreten (z.B. Netzwerkfehler, API-Fehler). - # Diese werden vom @retry_on_failure Decorator der search_company_article Methode behandelt. - # Werfen Sie die Exception erneut, damit der Decorator sie fangen kann. + validated_page = check_page(title) + if validated_page: + return validated_page except Exception as e_search: - self.logger.error(f"Fehler waehrend Wikipedia-Suche fuer '{term[:100]}...': {type(e_search).__name__} - {e_search}") # Gekuerzt loggen - # Werfen Sie die Exception erneut - raise e_search + self.logger.error( + f"Fehler waehrend Wikipedia-Suche fuer '{term[:100]}...': " + f"{type(e_search).__name__} - {e_search}" + ) + raise e_search + self.logger.warning( + f"Kein passender & validierter Wikipedia-Artikel fuer '{company_name[:100]}...' gefunden nach Pruefung aller Begriffe und Optionen." + ) + return None - # Wenn alle Suchbegriffe und alle Ergebnisse geprueft wurden und kein validierter Artikel gefunden wurde - self.logger.warning(f"Kein passender & validierter Wikipedia-Artikel fuer '{company_name[:100]}...' gefunden nach Pruefung aller Begriffe und Optionen.") # Gekuerzt loggen - return None # Gebe None zurueck, signalisiert, dass kein passender Artikel gefunden wurde - - - @retry_on_failure # Wende den Decorator an, da es externe Calls macht (Requests/BeautifulSoup) + @retry_on_failure def extract_company_data(self, page_url): """ Extrahiert Firmendaten (erster Absatz, Infobox-Werte, Kategorien) von einer gegebenen Wikipedia-Artikel-URL. - - Args: - page_url (str): Die URL des Wikipedia-Artikels. - - Returns: - dict: Ein Dictionary mit den extrahierten Daten oder Default-Werten ('k.A.'). - Formate fuer Umsatz/Mitarbeiter sind Strings ("123" oder "k.A."). - Wirft Exception bei API/Netzwerk-Fehlern nach Retries (von _get_page_soup). """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist - # Default-Ergebnis im Fehlerfall oder bei ungueltiger URL - default_result = {'url': page_url if page_url else 'k.A.', 'first_paragraph': 'k.A.', 'branche': 'k.A.', 'umsatz': 'k.A.', 'mitarbeiter': 'k.A.', 'categories': 'k.A.'} + default_result = { + 'url': page_url if page_url else 'k.A.', + 'first_paragraph': 'k.A.', + 'branche': 'k.A.', + 'umsatz': 'k.A.', + 'mitarbeiter': 'k.A.', + 'categories': 'k.A.' + } - # Grundlegende URL-Pruefung - # Stelle sicher, dass es ein String ist und wie eine Wikipedia-URL aussieht. if not page_url or not isinstance(page_url, str) or "wikipedia.org/wiki/" not in page_url.lower(): - self.logger.warning(f"extract_company_data: Ungueltige oder keine Wikipedia-URL '{page_url[:100]}...'.") # Gekuerzt loggen - return default_result # Gebe Default-Ergebnis zurueck + self.logger.warning( + f"extract_company_data: Ungueltige oder keine Wikipedia-URL '{page_url[:100]}...'." + ) + return default_result - - self.logger.info(f"Extrahiere Daten fuer Wiki-URL: {page_url[:100]}...") # Logge den Start der Extraktion (gekuerzt) - - # Hole das Soup-Objekt der Seite (nutzt interne Methode mit Retry). - # _get_page_soup wirft Exception bei endgueltigem Fehler, die hier nicht gefangen wird (weitergereicht). + self.logger.info(f"Extrahiere Daten fuer Wiki-URL: {page_url[:100]}...") soup = self._get_page_soup(page_url) - # Wenn _get_page_soup None zurueckgegeben hat (z.B. wegen ungueltiger URL, die am Anfang nicht gefiltert wurde) if not soup: - self.logger.error(f" -> Fehler: Konnte Seite {page_url[:100]}... nicht laden oder parsen.") # Gekuerzt loggen - # Das default_result enthaelt bereits die URL und k.A. fuer Daten. - return default_result # Gebe Default-Ergebnis zurueck + self.logger.error(f" -> Fehler: Konnte Seite {page_url[:100]}... nicht laden oder parsen.") + return default_result - - # --- Extrahiere die einzelnen Datenpunkte aus dem Soup-Objekt --- self.logger.debug(" -> Extrahiere erster Absatz...") - first_paragraph = self._extract_first_paragraph_from_soup(soup) # Nutzt interne Methode + first_paragraph = self._extract_first_paragraph_from_soup(soup) self.logger.debug(" -> Extrahiere Kategorien...") - categories_val = self.extract_categories(soup) # Nutzt interne Methode + categories_val = self.extract_categories(soup) self.logger.debug(" -> Extrahiere Branche aus Infobox...") - # Nutzt interne Methode _extract_infobox_value, die globale extract_numeric_value nutzt (Block 5) branche_val = self._extract_infobox_value(soup, 'branche') self.logger.debug(" -> Extrahiere Umsatz aus Infobox...") - # Nutzt interne Methode _extract_infobox_value, die globale extract_numeric_value nutzt (Block 5) umsatz_val = self._extract_infobox_value(soup, 'umsatz') self.logger.debug(" -> Extrahiere Mitarbeiter aus Infobox...") - # Nutzt interne Methode _extract_infobox_value, die globale extract_numeric_value nutzt (Block 5) mitarbeiter_val = self._extract_infobox_value(soup, 'mitarbeiter') - - # Baue das Ergebnis-Dictionary zusammen result = { - 'url': page_url, # Die URL, von der extrahiert wurde + 'url': page_url, 'first_paragraph': first_paragraph, 'branche': branche_val, 'umsatz': umsatz_val, @@ -3544,11 +3201,12 @@ class WikipediaScraper: 'categories': categories_val } - # Loggen Sie eine Zusammenfassung der extrahierten Daten (gekuerzt) - self.logger.info(f" -> Extrahierte Daten: P='{first_paragraph[:50]}...', B='{branche_val}', U='{umsatz_val}', M='{mitarbeiter_val}', C='{categories_val[:50]}...'") - - return result # Gebe das Dictionary mit den extrahierten Daten zurueck - + self.logger.info( + f" -> Extrahierte Daten: P='{first_paragraph[:50]}...', " + f"B='{branche_val}', U='{umsatz_val}', M='{mitarbeiter_val}', " + f"C='{categories_val[:50]}...'" + ) + return result # ============================================================================== # Ende Handler Klassen Block @@ -3565,7 +3223,7 @@ class DataProcessor: Zeilen sowie die Steuerung verschiedener Batch-Modi und Dienstprogramme. Nutzt Instanzen von Handler-Klassen (Sheet, Wiki etc.) als Worker. """ - def __init__(self, sheet_handler, wiki_scraper): # Akzeptiert benoetigte Worker-Instanzen + def __init__(self, sheet_handler, wiki_scraper): """ Initialisiert den DataProcessor mit Instanzen von Handler-Klassen. @@ -3581,13 +3239,13 @@ class DataProcessor: # Ueberpruefen Sie, ob gueltige Handler-Instanzen uebergeben wurden if not isinstance(sheet_handler, GoogleSheetHandler): - # Logge einen kritischen Fehler und werfe eine Exception - self.logger.critical("DataProcessor Init FEHLER: Kein gueltiger GoogleSheetHandler uebergeben!") - raise ValueError("DataProcessor benoetigt eine gueltige GoogleSheetHandler Instanz.") + # Logge einen kritischen Fehler und werfe eine Exception + self.logger.critical("DataProcessor Init FEHLER: Kein gueltiger GoogleSheetHandler uebergeben!") + raise ValueError("DataProcessor benoetigt eine gueltige GoogleSheetHandler Instanz.") if not isinstance(wiki_scraper, WikipediaScraper): - # Logge einen kritischen Fehler und werfe eine Exception - self.logger.critical("DataProcessor Init FEHLER: Kein gueltiger WikipediaScraper uebergeben!") - raise ValueError("DataProcessor benoetigt eine gueltige WikipediaScraper Instanz.") + # Logge einen kritischen Fehler und werfe eine Exception + self.logger.critical("DataProcessor Init FEHLER: Kein gueltiger WikipediaScraper uebergeben!") + raise ValueError("DataProcessor benoetigt eine gueltige WikipediaScraper Instanz.") # Speichern Sie die Handler-Instanzen als Attribute der Instanz self.sheet_handler = sheet_handler @@ -3595,40 +3253,29 @@ class DataProcessor: # self.openai_handler = openai_handler # Beispiel, falls ausgelagert # self.serpapi_handler = serpapi_handler # Beispiel, falls ausgelagert - # Attribute fuer ML-Modellierung (werden beim ersten Bedarf geladen in _load_ml_model) # Initialisieren Sie diese mit None self.model = None self.imputer = None - self._expected_features = None # Liste der erwarteten Feature-Spalten fuer Vorhersage - + self._expected_features = None # Liste der erwarteten Feature-Spalten fuer Vorhersage self.logger.info("DataProcessor initialisiert mit Handlern.") - # Definieren Sie hier (oder als Klassenattribut) die Zuordnung von Schritt-Typen # zu den relevanten Spaltenschluesseln fuer die Statuspruefung. # Diese werden von _should_run_based_on_status (Block 18) verwendet. - # HINWEIS: Die Logik, ob ein Schritt ausgefuehrt werden soll, ist komplexer - # als nur ein Timestamp (z.B. 'find_wiki_serp' braucht auch leere M und Groesse; - # 'summarize_website' braucht gefuellt AR). Die Methode _should_run_based_on_status - # wird hauptsaechlich fuer die sequentielle Verarbeitung (_process_single_row) - # und den Re-Eval Modus verfeinert. Batch-Modi haben oft ihre eigene spezifische - # Logik zur Zeilenauswahl, die im jeweiligen Batch-Methoden implementiert ist. - # Diese Map ist primaer fuer die Logik in _should_run_based_on_status (Block 18) relevant. self._step_status_map = { - 'wiki_verify': "Wiki Verif. Timestamp", # AX - Trigger fuer Wiki-Verifikation (S-Y) - 'website_scrape': "Website Scrape Timestamp", # AT - Trigger fuer Website-Scraping (AR, AS) - 'summarize_website': "Website Scrape Timestamp", # AT - Trigger fuer Website-Summarization (AS) - 'branch_eval': "Timestamp letzte Pruefung", # AO - Trigger fuer Branchen-Evaluation (W-Y) - 'find_wiki_serp': "SerpAPI Wiki Search Timestamp", # AY - Trigger fuer SerpAPI Wiki Search (M, AY) - 'contact_search': "Contact Search Timestamp", # AM - Trigger fuer LinkedIn Kontakt Suche (AI-AL, AM) - 'wiki_updates_from_chatgpt': "Chat Wiki Konsistenzpruefung", # S - Trigger fuer U->M Uebernahme (S) - # 'wiki_extract': "Wikipedia Timestamp", # AN - Trigger fuer Wiki-Extraktion (M-R, AN) - Wird in _process_single_row speziell geprueft - # 'ml_predict': "Geschaetzter Techniker Bucket" # AU - Trigger fuer ML Schaetzung (AU) - Wird in _process_single_row oder separater Methode speziell geprueft + 'wiki_verify': "Wiki Verif. Timestamp", + 'website_scrape': "Website Scrape Timestamp", + 'summarize_website': "Website Scrape Timestamp", + 'branch_eval': "Timestamp letzte Pruefung", + 'find_wiki_serp': "SerpAPI Wiki Search Timestamp", + 'contact_search': "Contact Search Timestamp", + 'wiki_updates_from_chatgpt': "Chat Wiki Konsistenzpruefung", + # 'wiki_extract': "Wikipedia Timestamp", # Speziell in _process_single_row behandelt + # 'ml_predict': "Geschaetzter Techniker Bucket" # Speziell in eigener Methode behandelt } - # --- Interne Hilfsmethode zur Statuspruefung einer Zelle --- # Dient zum sicheren Abrufen von Werten aus einer Zeile unter Verwendung von COLUMN_MAP. # Nutzt globale Helfer: COLUMN_MAP, logger. @@ -3644,26 +3291,26 @@ class DataProcessor: Returns: str: Der Wert der Zelle als String, oder '' wenn nicht verfuegbar. """ - # Verwenden Sie logger, da das Logging jetzt konfiguriert ist # Ermitteln Sie den Index der Spalte aus COLUMN_MAP (Block 1) idx = COLUMN_MAP.get(column_key) # Pruefen Sie, ob der Schluessel in COLUMN_MAP gefunden wurde if idx is None: - # Logge einen Fehler, aber geben Sie einen leeren String zurueck, um Abstuerze zu vermeiden - self.logger.error(f"_get_cell_value_safe: Schluessel '{column_key}' nicht in COLUMN_MAP gefunden.") - return '' # Gebe leeren String zurueck, wenn Schluessel fehlt - + # Logge einen Fehler, aber gebe einen leeren String zurueck + self.logger.error(f"_get_cell_value_safe: Schluessel '{column_key}' nicht in COLUMN_MAP gefunden.") + return '' # Gebe leeren String zurueck, wenn Schluessel fehlt # Pruefen Sie, ob die Zeile lang genug ist, um auf diesen Index zuzugreifen if len(row) > idx: # Rueckgabe des Wertes, sicherstellen, dass es nicht None ist return row[idx] if row[idx] is not None else '' else: - # Logge auf Debug-Level, wenn der Index existiert, aber die Zeile zu kurz ist. - # Dies kann auf inkonsistente Zeilenlaengen im Sheet hindeuten. - self.logger.debug(f"_get_cell_value_safe: Index {idx} fuer '{column_key}' ist gueltig, aber Zeile ist zu kurz (Laenge {len(row)}). Gebe leeren String zurueck.") - return '' # Gebe leeren String zurueck, wenn die Zeile zu kurz ist + # Logge auf Debug-Level, wenn der Index existiert, aber die Zeile zu kurz ist. + self.logger.debug( + f"_get_cell_value_safe: Index {idx} fuer '{column_key}' ist gueltig, " + f"aber Zeile ist zu kurz (Laenge {len(row)}). Gebe leeren String zurueck." + ) + return '' # Gebe leeren String zurueck, wenn die Zeile zu kurz ist # Der Code sollte niemals hierher gelangen. # return '' # Fallback Rueckgabe @@ -11359,7 +11006,7 @@ def main(): try: # Der GoogleSheetHandler Init (_init_ Methode) baut die Verbindung auf und laedt Daten. # Fehler werden dort gefangen und als ConnectionError erneut geworfen. - sheet_handler = GoogleSheetHandler() + sheet_handler = GoogleSheetHandler() #<- Zeile 11362 logger.info("GoogleSheetHandler erfolgreich initialisiert.") except ConnectionError as e: # Wenn die Initialisierung des SheetHandlers fehlschlaegt (Verbindungs-/Ladefehler) @@ -13593,7 +13240,7 @@ def main(): try: # Der GoogleSheetHandler Init (_init_ Methode) baut die Verbindung auf und laedt Daten. # Fehler werden dort gefangen und als ConnectionError erneut geworfen. - sheet_handler = GoogleSheetHandler() + sheet_handler = GoogleSheetHandler() #<- Zeile 13596 logger.info("GoogleSheetHandler erfolgreich initialisiert.") except ConnectionError as e: # Wenn die Initialisierung des SheetHandlers fehlschlaegt (Verbindungs-/Ladefehler)