From d728c4c590054aaa85de30004aecbb2ac7c462f9 Mon Sep 17 00:00:00 2001 From: Floke Date: Tue, 8 Jul 2025 10:21:13 +0000 Subject: [PATCH] bugfix --- dealfront_enrichment.py | 160 +++++++++++++++------------------------- 1 file changed, 58 insertions(+), 102 deletions(-) diff --git a/dealfront_enrichment.py b/dealfront_enrichment.py index 53f83379..7d4ed440 100644 --- a/dealfront_enrichment.py +++ b/dealfront_enrichment.py @@ -15,25 +15,29 @@ from config import Config, DEALFRONT_LOGIN_URL, DEALFRONT_CREDENTIALS_FILE, DEAL OUTPUT_DIR = "/app/output" # Logging-Konfiguration -LOG_LEVEL = logging.DEBUG if Config.DEBUG else logging.INFO +LOG_LEVEL = logging.INFO # Wir setzen den Standard auf INFO für eine saubere Ausgabe LOG_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s' -# Root-Logger konfigurieren (gilt für alle Bibliotheken) -logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True) +# Root-Logger für die Konsolenausgabe konfigurieren +logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True, handlers=[logging.StreamHandler()]) -# Logger für Selenium auf INFO setzen, um Spam zu vermeiden -logging.getLogger("selenium").setLevel(logging.INFO) +# Selenium-Logger auf WARNING setzen, um den Spam zu unterdrücken +logging.getLogger("selenium").setLevel(logging.WARNING) -# Logger für unser eigenes Skript +# Eigener Logger für unser Skript logger = logging.getLogger(__name__) -# FileHandler hinzufügen, um in eine Datei zu loggen -log_filename = f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.log" +# FileHandler hinzufügen, um ALLES (inkl. DEBUG) in eine .txt-Datei zu schreiben +# Wichtig: Dieser Handler hat sein eigenes Level (DEBUG), um alles zu erfassen. +log_filename = f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.txt" log_filepath = os.path.join(OUTPUT_DIR, log_filename) file_handler = logging.FileHandler(log_filepath, mode='w', encoding='utf-8') +file_handler.setLevel(logging.DEBUG) # Alles in die Datei schreiben file_handler.setFormatter(logging.Formatter(LOG_FORMAT)) logging.getLogger().addHandler(file_handler) +logger.info(f"Logging konfiguriert. Konsolenausgabe auf Level {logging.getLevelName(LOG_LEVEL)}. Log-Datei: {log_filepath}") + OUTPUT_DIR = "/app/output" class DealfrontScraper: @@ -108,77 +112,52 @@ class DealfrontScraper: except Exception as e: logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}") - def login_and_navigate_to_target(self): + def login_and_find_list(self, search_name): """ - Führt den Login durch und klickt dann auf dem Dashboard auf den - "Quick Link", um zur Target-Seite zu gelangen. + Führt den gesamten Prozess vom Login bis zum Laden der Zielliste robust aus. """ - if not self.driver: return False - username, password = self._load_credentials() - if not username or not password: return False - try: - # SCHRITT 1: LOGIN + # === LOGIN === logger.info(f"Navigiere zur Login-Seite: {DEALFRONT_LOGIN_URL}") self.driver.get(DEALFRONT_LOGIN_URL) - - self.wait.until(EC.visibility_of_element_located((By.NAME, "email"))).send_keys(username) - self.driver.find_element(By.CSS_SELECTOR, "input[type='password']").send_keys(password) + self.wait.until(EC.visibility_of_element_located((By.NAME, "email"))).send_keys(self.username) + self.driver.find_element(By.CSS_SELECTOR, "input[type='password']").send_keys(self.password) self.driver.find_element(By.XPATH, "//button[normalize-space()='Log in']").click() logger.info("Login-Befehl gesendet.") - # SCHRITT 2: AUF QUICK-LINK-KACHEL WARTEN UND KLICKEN - logger.info("Warte auf Dashboard und 'Prospects finden'-Link in den Quick Links.") - - # Dieser XPath ist sehr spezifisch für den Link in der "Quick links"-Kachel + # === NAVIGATION ZUM TARGET BEREICH === + logger.info("Warte auf Dashboard und den 'Prospects finden' Quick-Link...") + # Wir warten geduldig (bis zu 30s) auf die Kachel, die uns zur Target-Seite bringt prospects_link_selector = (By.XPATH, "//a[@data-test-target-product-tile]") prospects_link = self.wait.until(EC.element_to_be_clickable(prospects_link_selector)) - - logger.info("'Prospects finden'-Link gefunden. Klicke darauf...") prospects_link.click() + logger.info("'Prospects finden' geklickt. Navigiere zur Target-Seite.") - # SCHRITT 3: NAVIGATION VERIFIZIEREN - logger.info("Warte auf die finale Target-Seite...") - verification_target_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']") - self.wait.until(EC.visibility_of_element_located(verification_target_selector)) + # === LADEN DER SPEZIFISCHEN SUCHE === + logger.info(f"Warte auf die Liste der Suchen und klicke auf '{search_name}'...") + search_item_selector = (By.XPATH, f"//div[contains(@class, 'truncate') and normalize-space()='{search_name}']") + search_item = self.wait.until(EC.element_to_be_clickable(search_item_selector)) + search_item.click() - logger.info("'Target'-Seite erfolgreich und vollständig geladen.") + # === VERIFIZIERUNG UND WARTEN AUF TABELLENDATEN === + logger.info(f"Suche '{search_name}' geladen. Warte auf das Rendern der Ergebnistabelle.") + table_header_selector = (By.XPATH, "//th[normalize-space()='Firma']") + self.wait.until(EC.visibility_of_element_located(table_header_selector)) + time.sleep(5) # Finale, großzügige Pause für das Laden der Tabellen-Daten via JS + + logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.") return True + except Exception as e: + logger.critical(f"Der Prozess ist fehlgeschlagen: {type(e).__name__}", exc_info=True) + self._save_debug_artifacts() # Speichert Screenshot und HTML für die Analyse + return False + except Exception as e: logger.critical(f"Login- oder Navigationsprozess fehlgeschlagen: {type(e).__name__}", exc_info=True) self._save_debug_artifacts() return False - def navigate_to_target(self): - """ - Navigiert zum 'Target'-Bereich und verifiziert den Erfolg in drei Schritten. - Dieser Ansatz ist maximal robust gegen Timing-Probleme von SPAs. - """ - try: - # SCHRITT 1: Befehl zur Navigation geben - logger.info(f"Gebe Navigationsbefehl zur Target-URL: {Config.DEALFRONT_TARGET_URL}") - self.driver.get(Config.DEALFRONT_TARGET_URL) - - # SCHRITT 2: Warten, bis die URL in der Adresszeile sich tatsächlich ändert. - url_part_to_wait_for = "/t/prospector/" - logger.info(f"Warte, bis die Browser-URL '{url_part_to_wait_for}' enthält...") - self.wait.until(EC.url_contains(url_part_to_wait_for)) - logger.info(f"URL-Wechsel bestätigt. Aktuelle URL: {self.driver.current_url}") - - # SCHRITT 3: ERST JETZT auf ein sichtbares Element auf der neuen Seite warten. - verification_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']") - logger.info(f"Warte auf Sichtbarkeit des Verifizierungs-Elements auf der Target-Seite: {verification_selector}") - self.wait.until(EC.visibility_of_element_located(verification_selector)) - - logger.info("'Target'-Seite erfolgreich und vollständig geladen.") - return True - - except Exception as e: - logger.critical(f"Navigation zur 'Target'-Seite endgültig fehlgeschlagen: {type(e).__name__}", exc_info=True) - self._save_debug_artifacts() - return False - def handle_overlays(self): """Sucht nach bekannten Popups/Overlays und schließt sie.""" try: @@ -204,36 +183,6 @@ class DealfrontScraper: except Exception as e: logger.warning(f"Fehler beim Schließen des Overlays, ignoriere und fahre fort: {e}") self._save_debug_artifacts() - - - def load_search(self, search_name): - """Lädt eine vordefinierte Suche anhand ihres Namens.""" - try: - logger.info(f"Warte, bis die Liste der gespeicherten Suchen sichtbar ist...") - # Wir warten auf den Container, der alle Such-Items enthält. - # Ein guter Selektor ist das `ul`-Element mit der ID, die mit `eb-popup` beginnt. - # Oder allgemeiner: Ein Element, das den Text einer bekannten Suche enthält. - search_list_container_selector = (By.XPATH, f"//div[normalize-space()='{search_name}']") - self.wait.until(EC.visibility_of_element_located(search_list_container_selector)) - logger.info(f"Liste gefunden. Klicke jetzt auf die vordefinierte Suche: '{search_name}'") - - # Der XPath sucht nach einem Element (div, a, button), dessen Text exakt übereinstimmt. - search_link_selector = (By.XPATH, f"//*[normalize-space()='{search_name}']") - search_link = self.wait.until(EC.element_to_be_clickable(search_link_selector)) - search_link.click() - - # Verifizieren, dass die Ergebnisse geladen sind, indem wir auf die Ergebnistabelle warten. - results_table_header_selector = (By.XPATH, "//th[normalize-space()='Firma']") - self.wait.until(EC.visibility_of_element_located(results_table_header_selector)) - logger.info(f"Suche '{search_name}' erfolgreich geladen und Ergebnisse angezeigt.") - - time.sleep(3) # Feste Wartezeit, damit alle Daten nachgeladen sind. - return True - - except Exception as e: - logger.critical(f"Laden der Suche '{search_name}' fehlgeschlagen: {type(e).__name__}", exc_info=True) - self._save_debug_artifacts() - return False def extract_current_page_results(self): """ @@ -298,7 +247,7 @@ class DealfrontScraper: self.driver.quit() if __name__ == "__main__": - logger.info("Starte Dealfront Automatisierung - Finaler Workflow") + logger.info("Starte Dealfront Automatisierung - Finaler Durchbruchsversuch") scraper = None try: @@ -306,35 +255,42 @@ if __name__ == "__main__": if not scraper.driver: raise Exception("WebDriver konnte nicht initialisiert werden.") - # === EINZIGER AUFRUF FÜR LOGIN UND NAVIGATION === - if not scraper.login_and_navigate_to_target(): - raise Exception("Login und Navigation fehlgeschlagen.") - - # Ab hier geht es weiter wie geplant... - if not scraper.load_search(Config.TARGET_SEARCH_NAME): - raise Exception(f"Laden der Suche '{Config.TARGET_SEARCH_NAME}' fehlgeschlagen.") - - scraper.handle_overlays() + # Ein einziger, robuster Aufruf für den gesamten Prozess bis zur Anzeige der Ergebnistabelle + if not scraper.login_and_find_list(Config.TARGET_SEARCH_NAME): + # Der Fehler wird bereits innerhalb der Methode geloggt und Artefakte werden gespeichert. + # Wir müssen hier nur noch den Prozess beenden. + raise Exception("Der Prozess vom Login bis zum Laden der Liste ist fehlgeschlagen. Details siehe Log.") + # Wenn wir hier ankommen, sind wir garantiert auf der richtigen Seite. + # Jetzt extrahieren wir die Daten. companies = scraper.extract_current_page_results() + + # Saubere Ausgabe der Ergebnisse in einer Tabelle if companies: df = pd.DataFrame(companies) + + # Pandas-Optionen für eine vollständige, ungeschnittene Ausgabe pd.set_option('display.max_rows', None) pd.set_option('display.max_columns', None) pd.set_option('display.width', 1000) pd.set_option('display.max_colwidth', None) + print("\n" + "="*80) print(" EXTRAHIERTE FIRMEN (ERSTE SEITE) ".center(80, "=")) print("="*80) print(df.to_string(index=False)) print("="*80 + "\n") - else: - logger.warning("Keine Firmen auf der ersten Seite extrahiert oder gefunden.") + + logger.info(f"{len(df)} Firmen erfolgreich extrahiert und in der Konsole ausgegeben.") - logger.info("Test erfolgreich abgeschlossen. Warte vor dem Schließen...") + else: + logger.warning("Obwohl die Seite geladen wurde, konnten keine Firmen extrahiert werden. Bitte HTML-Dump prüfen.") + + logger.info("Phase 2a Test erfolgreich abgeschlossen. Warte vor dem Schließen...") time.sleep(10) except Exception as e: + # Hier fangen wir nur noch die Exceptions aus dem __main__-Block selbst logger.critical(f"Ein kritischer Fehler ist im Hauptprozess aufgetreten: {e}", exc_info=False) finally: if scraper: