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-08 12:28:01 +00:00
parent e465059df8
commit df1daa5932
1 changed files with 38 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
dealfront_enrichment.py
View File

@@ -85,45 +85,45 @@ 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 Laden der Zielliste robust aus, Führt den gesamten Prozess vom Login bis zum Laden der Zielliste robust aus,
indem auf das Erscheinen der ersten Datenzeile gewartet wird. indem auf das Erscheinen der ersten Datenzeile gewartet wird.
""" """
try: try:
# === SCHRITTE 1-3: LOGIN, NAVIGATION, SUCHE LADEN (sind bereits stabil) === # === SCHRITTE 1-3: LOGIN, NAVIGATION, SUCHE LADEN (sind bereits stabil) ===
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.") logger.info("Login-Befehl gesendet.")
logger.info("Warte auf Dashboard und den 'Prospects finden' 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.") logger.info("'Prospects finden' geklickt.")
logger.info(f"Warte auf die Liste der Suchen und klicke auf '{search_name}'...") 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}']")
search_item = self.wait.until(EC.element_to_be_clickable(search_item_selector)) search_item = self.wait.until(EC.element_to_be_clickable(search_item_selector))
search_item.click() search_item.click()
# === SCHRITT 4: FINALE, ROBUSTE VERIFIZIERUNG === # === SCHRITT 4: FINALE, ROBUSTE VERIFIZIERUNG ===
logger.info(f"Suche '{search_name}' geladen. Warte auf die erste Datenzeile in der Tabelle...") logger.info(f"Suche '{search_name}' geladen. Warte auf die erste Datenzeile in der Tabelle...")
# Wir warten auf das Erscheinen des Links in der ERSTEN Datenzeile. # Wir warten auf das Erscheinen des Links in der ERSTEN Datenzeile.
# Dies ist der Beweis, dass die API-Abfrage abgeschlossen und die Tabelle gerendert wurde. # Dies ist der Beweis, dass die API-Abfrage abgeschlossen und die Tabelle gerendert wurde.
first_company_link_selector = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id] .sticky-column a.t-highlight-text") first_company_link_selector = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id] .sticky-column a.t-highlight-text")
self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, first_company_link_selector))) self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, first_company_link_selector)))
logger.info("Erste Datenzeile erfolgreich gefunden. Die Seite ist bereit für die Extraktion.") logger.info("Erste Datenzeile erfolgreich gefunden. Die Seite ist bereit für die Extraktion.")
return True return True
except Exception as e: except Exception as e:
logger.critical(f"Der Prozess ist fehlgeschlagen: {type(e).__name__}", exc_info=True) logger.critical(f"Der Prozess ist fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts() self._save_debug_artifacts()
return False return False
def extract_current_page_results(self): def extract_current_page_results(self):
""" """