Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dealfront_enrichment.py aktualisiert

Browse Source
This commit is contained in:
Floke
2025-07-10 13:23:25 +00:00
parent d16b503e3e
commit 3183a7f796
1 changed files with 49 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
dealfront_enrichment.py
View File

@@ -84,29 +84,40 @@ class DealfrontScraper:
logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}") logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}")
def login_and_find_list(self, search_name): def login_and_find_list(self, search_name):
"""
Führt den gesamten Prozess vom Login bis zum Auffinden der Suchergebnisliste durch.
Diese Methode ist für Geschwindigkeit und Stabilität optimiert.
"""
try: try:
# Schritt 1: Login
logger.info(f"Navigiere zur Login-Seite: {TempConfig.DEALFRONT_LOGIN_URL}") logger.info(f"Navigiere zur Login-Seite: {TempConfig.DEALFRONT_LOGIN_URL}")
self.driver.get(TempConfig.DEALFRONT_LOGIN_URL) self.driver.get(TempConfig.DEALFRONT_LOGIN_URL)
self.wait.until(EC.visibility_of_element_located((By.NAME, "email"))).send_keys(self.username) 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.CSS_SELECTOR, "input[type='password']").send_keys(self.password)
self.driver.find_element(By.XPATH, "//button[normalize-space()='Log in']").click() self.driver.find_element(By.XPATH, "//button[normalize-space()='Log in']").click()
logger.info("Login-Befehl gesendet. Warte auf Navigation...") logger.info("Login-Befehl gesendet.")
# Schritt 2: Navigation zur Target-Seite via Klick auf den Quick-Link
logger.info("Warte auf Dashboard und den 'Prospects finden' Quick-Link...")
prospects_link_selector = (By.XPATH, "//a[@data-test-target-product-tile]") prospects_link_selector = (By.XPATH, "//a[@data-test-target-product-tile]")
prospects_link = self.wait.until(EC.element_to_be_clickable(prospects_link_selector)) prospects_link = self.wait.until(EC.element_to_be_clickable(prospects_link_selector))
prospects_link.click() prospects_link.click()
logger.info("'Prospects finden' geklickt. Lade Suchliste...") logger.info("'Prospects finden' geklickt.")
# Schritt 3: Vordefinierte Suche laden
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_selector = (By.XPATH, f"//div[contains(@class, 'truncate') and normalize-space()='{search_name}']")
self.wait.until(EC.element_to_be_clickable(search_item_selector)).click() self.wait.until(EC.element_to_be_clickable(search_item_selector)).click()
logger.info(f"Suche '{search_name}' geladen. Warte auf Ergebnisse.") # Schritt 4: Erfolg verifizieren durch Warten auf die Tabelle
logger.info(f"Suche '{search_name}' geladen. Warte auf das Rendern der Ergebnistabelle.")
first_row_locator = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id]") first_row_locator = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id]")
self.wait.until(EC.visibility_of_element_located(first_row_locator)) self.wait.until(EC.visibility_of_element_located(first_row_locator))
logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.") logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.")
return True return True
except Exception as e: except Exception as e:
logger.critical(f"Prozess bis zum Laden der Liste fehlgeschlagen: {type(e).__name__}", exc_info=True) logger.critical(f"Der Prozess bis zum Laden der Liste ist fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts() self._save_debug_artifacts()
return False return False
@@ -118,11 +129,8 @@ class DealfrontScraper:
logger.info("Extrahiere Ergebnisse mit dem finalen, präzisen Selektor...") logger.info("Extrahiere Ergebnisse mit dem finalen, präzisen Selektor...")
results = [] results = []
# 1. Warten, bis die erste Daten-Zelle (Firmenname) sichtbar ist. # 1. Da die vorige Methode schon gewartet hat, können wir hier direkt suchen.
first_company_link_selector = (By.CSS_SELECTOR, "td.sticky-after-checkbox a.t-highlight-text") # Der XPath findet NUR Zeilen, die einen Firmen-Link enthalten, und ignoriert Geister-Zeilen.
self.wait.until(EC.visibility_of_element_located(first_company_link_selector))
# 2. Finde NUR die Zeilen, die tatsächlich einen Firmennamen-Link enthalten.
rows_with_data_selector = (By.XPATH, "//table[@id='t-result-table']/tbody/tr[.//a[contains(@class, 't-highlight-text')]]") rows_with_data_selector = (By.XPATH, "//table[@id='t-result-table']/tbody/tr[.//a[contains(@class, 't-highlight-text')]]")
rows = self.driver.find_elements(*rows_with_data_selector) rows = self.driver.find_elements(*rows_with_data_selector)
logger.info(f"{len(rows)} gültige Datenzeilen zur Verarbeitung gefunden.") logger.info(f"{len(rows)} gültige Datenzeilen zur Verarbeitung gefunden.")
@@ -156,13 +164,11 @@ class DealfrontScraper:
self._save_debug_artifacts() self._save_debug_artifacts()
return [] return []
def close(self): def close(self):
if self.driver: if self.driver:
logger.info("Schließe den WebDriver.") logger.info("Schließe den WebDriver.")
self.driver.quit() self.driver.quit()
if __name__ == "__main__": if __name__ == "__main__":
logger.info("Starte Dealfront Automatisierung - DEBUG-MODUS") logger.info("Starte Dealfront Automatisierung - DEBUG-MODUS")
scraper = None scraper = None