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:54:13 +00:00
parent f4004ef11f
commit 2968586b46
1 changed files with 32 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
dealfront_enrichment.py
View File

@@ -84,39 +84,39 @@ 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):
try: """
# Schritt 1: Login Führt den gesamten Prozess vom Login bis zum Auffinden der Suchergebnisliste durch.
logger.info(f"Navigiere zur Login-Seite: {TempConfig.DEALFRONT_LOGIN_URL}") """
self.driver.get(TempConfig.DEALFRONT_LOGIN_URL) try:
self.wait.until(EC.visibility_of_element_located((By.NAME, "email"))).send_keys(self.username) logger.info(f"Navigiere zur Login-Seite: {TempConfig.DEALFRONT_LOGIN_URL}")
self.driver.find_element(By.CSS_SELECTOR, "input[type='password']").send_keys(self.password) self.driver.get(TempConfig.DEALFRONT_LOGIN_URL)
self.driver.find_element(By.XPATH, "//button[normalize-space()='Log in']").click() self.wait.until(EC.visibility_of_element_located((By.NAME, "email"))).send_keys(self.username)
logger.info("Login-Befehl gesendet.") 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.")
logger.info("Warte auf Dashboard und den 'Prospects finden' Quick-Link...")
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.click()
logger.info("'Prospects finden' geklickt.")
# Schritt 2: Navigation zur Target-Seite via Klick auf den Quick-Link logger.info(f"Warte auf die Liste der Suchen und klicke auf '{search_name}'...")
logger.info("Warte auf Dashboard und den 'Prospects finden' Quick-Link...") search_item_selector = (By.XPATH, f"//div[contains(@class, 'truncate') and normalize-space()='{search_name}']")
prospects_link_selector = (By.XPATH, "//a[@data-test-target-product-tile]") self.wait.until(EC.element_to_be_clickable(search_item_selector)).click()
prospects_link = self.wait.until(EC.element_to_be_clickable(prospects_link_selector))
prospects_link.click() logger.info(f"Suche '{search_name}' geladen. Warte auf Ergebnisse.")
logger.info("'Prospects finden' geklickt.") first_row_locator = (By.CSS_SELECTOR, "td.sticky-after-checkbox a.t-highlight-text")
self.wait.until(EC.visibility_of_element_located(first_row_locator))
# Schritt 3: Vordefinierte Suche laden
logger.info(f"Warte auf die Liste der Suchen und klicke auf '{search_name}'...") logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.")
search_item_selector = (By.XPATH, f"//div[contains(@class, 'truncate') and normalize-space()='{search_name}']") return True
self.wait.until(EC.element_to_be_clickable(search_item_selector)).click() except Exception as e:
logger.critical(f"Prozess bis zum Laden der Liste fehlgeschlagen: {type(e).__name__}", exc_info=True)
# Schritt 4: Erfolg verifizieren durch Warten auf die Tabelle self._save_debug_artifacts()
logger.info(f"Suche '{search_name}' geladen. Warte auf das Rendern der Ergebnistabelle.") return False
first_row_locator = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id]")
self.wait.until(EC.visibility_of_element_located(first_row_locator))
logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.")
return True
except Exception as e:
logger.critical(f"Der Prozess bis zum Laden der Liste ist fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts()
return False
# HIER IST DIE KORREKTE EINRÜCKUNG
def extract_current_page_results(self): def extract_current_page_results(self):
""" """
Extrahiert Daten mit einem schnellen, direkten Ansatz, der Geister-Zeilen ignoriert. Extrahiert Daten mit einem schnellen, direkten Ansatz, der Geister-Zeilen ignoriert.
@@ -125,29 +125,18 @@ class DealfrontScraper:
logger.info("Extrahiere Ergebnisse mit dem finalen, direkten Selektor-Ansatz...") logger.info("Extrahiere Ergebnisse mit dem finalen, direkten Selektor-Ansatz...")
results = [] results = []
# 1. Warten, bis die erste Daten-Zelle (Firmenname) sichtbar ist. # 1. Wir wissen, dass die Seite geladen ist, und können direkt suchen.
# Dies ist unser einziger Wartepunkt und bestätigt, dass die Daten geladen sind.
first_company_link_selector = (By.CSS_SELECTOR, "td.sticky-column a.t-highlight-text")
self.wait.until(EC.visibility_of_element_located(first_company_link_selector))
# 2. Finde ALLE Firmen-Links und ALLE Website-Links auf der Seite auf einmal.
# Das ist extrem schnell, da es nur zwei Suchbefehle an den Browser sind.
company_elements = self.driver.find_elements(By.CSS_SELECTOR, "td.sticky-column a.t-highlight-text") company_elements = self.driver.find_elements(By.CSS_SELECTOR, "td.sticky-column a.t-highlight-text")
website_elements = self.driver.find_elements(By.CSS_SELECTOR, "a.text-gray-400.t-highlight-text") website_elements = self.driver.find_elements(By.CSS_SELECTOR, "a.text-gray-400.t-highlight-text")
logger.info(f"{len(company_elements)} Firmennamen und {len(website_elements)} Webseiten-Elemente gefunden.") logger.info(f"{len(company_elements)} Firmennamen und {len(website_elements)} Webseiten-Elemente gefunden.")
# 3. Ordne die Ergebnisse anhand ihrer Reihenfolge im DOM zu.
# Wir gehen davon aus, dass die Anzahl übereinstimmt.
if not company_elements: if not company_elements:
logger.warning("Keine Firmen mit dem angegebenen Selektor gefunden.") logger.warning("Keine Firmen mit dem angegebenen Selektor gefunden.")
self._save_debug_artifacts()
return [] return []
for i in range(len(company_elements)): for i in range(len(company_elements)):
company_name = company_elements[i].get_attribute("title").strip() company_name = company_elements[i].get_attribute("title").strip()
# Wir ordnen die Webseite anhand des Indexes zu.
# Wenn es weniger Webseiten als Firmen gibt, fangen wir das ab.
website = website_elements[i].text.strip() if i < len(website_elements) else "N/A" website = website_elements[i].text.strip() if i < len(website_elements) else "N/A"
results.append({'name': company_name, 'website': website}) results.append({'name': company_name, 'website': website})