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-13 10:37:49 +00:00
parent 41ce9f7d1e
commit 97f4709d51
1 changed files with 50 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
dealfront_enrichment.py
View File

@@ -131,77 +131,68 @@ class DealfrontScraper:
return results return results
def scrape_all_pages(self, max_pages=10): def scrape_all_pages(self, max_pages=6):
""" """
Iteriert durch alle Ergebnisseiten und sammelt ALLE Firmenzeilen (auch Duplikate). Iteriert durch alle Ergebnisseiten, scrollt auf jeder Seite nach unten,
um alle Daten zu laden, und extrahiert sie dann. Verhindert Duplikate.
""" """
all_companies = [] all_companies = {} # Verwende ein Dictionary, um Duplikate zu vermeiden
for page_number in range(1, max_pages + 1): for page_number in range(1, max_pages + 1):
logger.info(f"--- Verarbeite Seite {page_number} ---") logger.info(f"--- Verarbeite Seite {page_number} ---")
self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "table#t-result-table")))
# Extrahiere aktuelle Seite # Warten, bis die Tabelle grundsätzlich geladen ist
table_selector = (By.CSS_SELECTOR, "table#t-result-table")
self.wait.until(EC.visibility_of_element_located(table_selector))
# --- NEU: Intelligente Scroll-Schleife ---
logger.info("Scrolle Seite nach unten, um alle Firmen zu laden...")
last_height = self.driver.execute_script("return document.querySelector('.scroll-viewport').scrollHeight")
while True:
self.driver.execute_script("document.querySelector('.scroll-viewport').scrollTo(0, document.querySelector('.scroll-viewport').scrollHeight);")
time.sleep(2) # Wartezeit für das Nachladen der Inhalte
new_height = self.driver.execute_script("return document.querySelector('.scroll-viewport').scrollHeight")
if new_height == last_height:
break
last_height = new_height
logger.info("Seitenende erreicht, alle Zeilen sollten geladen sein.")
# Extrahiere die Daten von der komplett geladenen Seite
page_results = self.extract_current_page_results() page_results = self.extract_current_page_results()
if not page_results and page_number > 1: if not page_results:
logger.warning("Keine Ergebnisse auf dieser Seite gefunden, breche ab.") logger.warning("Konnte auf dieser Seite keine Ergebnisse extrahieren, obwohl gescrollt wurde.")
break
logger.info(f"Seite {page_number}: {len(page_results)} Firmen gefunden. Gesamt bisher: {len(all_companies) + len(page_results)}") for company in page_results:
unique_key = (company.get('name'), company.get('website'))
if unique_key not in all_companies:
all_companies[unique_key] = company
# ALLE Ergebnisse einfach anhängen (keine Deduplizierung!) logger.info(f"Seite {page_number}: {len(page_results)} Firmen gefunden. Gesamt einzigartig: {len(all_companies)}")
all_companies.extend(page_results)
# --- Pagination-Button-Logik bleibt wie gehabt --- # --- Paginierungs-Logik ---
try: try:
pagination_nav = self.driver.find_element(By.CSS_SELECTOR, "nav.eb-pagination") # Wir merken uns die ID der LETZTEN Zeile, um den Wechsel zu verifizieren
buttons = pagination_nav.find_elements(By.CSS_SELECTOR, "a.eb-pagination-button") last_row_id = self.driver.find_element(By.XPATH, "(//table[@id='t-result-table']/tbody/tr[@id])[last()]").get_attribute("id")
logger.info(f"Gefundene Paginierungs-Buttons auf Seite {page_number}: {len(buttons)}")
for idx, btn in enumerate(buttons):
btn_text = btn.text.strip()
btn_classes = btn.get_attribute('class')
btn_html = btn.get_attribute('outerHTML')
has_svg = "svg" in btn_html
logger.info(f"Button {idx}: Text='{btn_text}', Klassen='{btn_classes}', SVG={has_svg}, HTML={btn_html}")
except Exception as e:
logger.warning(f"Konnte Pagination-Buttons nicht vollständig loggen: {e}")
buttons = []
# Suche nach dem Weiter-Button next_button_selector = (By.XPATH, "//nav[contains(@class, 'eb-pagination')]//a[not(contains(@class, 'disabled'))][last()]")
next_button = None next_button = self.driver.find_element(*next_button_selector)
for btn in buttons:
btn_html = btn.get_attribute('outerHTML') if "fa-angle-right" not in next_button.get_attribute('innerHTML'):
if ("svg" in btn_html) and ("disabled" not in btn.get_attribute('class')) and (btn.text.strip() == ""): logger.info("Letzte Seite erreicht (kein 'Weiter'-Pfeil im letzten Button).")
next_button = btn
break break
if not next_button: self.driver.execute_script("arguments[0].click();", next_button)
logger.info(f"Klicke auf 'Weiter' zu Seite {page_number + 1}...")
# Warte darauf, dass die letzte Zeile der alten Seite verschwindet
self.wait.until(EC.staleness_of(self.driver.find_element(By.ID, last_row_id)))
logger.info("Seitenwechsel erfolgreich verifiziert.")
except (NoSuchElementException, TimeoutException):
logger.info("Kein klickbarer 'Weiter'-Button mehr gefunden. Paginierung abgeschlossen.") logger.info("Kein klickbarer 'Weiter'-Button mehr gefunden. Paginierung abgeschlossen.")
break break
logger.info("Klicke auf 'Weiter'-Button...") return list(all_companies.values())
try:
# Vor dem Klick: erste Zeile merken
try:
first_row_element = self.driver.find_element(By.CSS_SELECTOR, "table#t-result-table tbody tr[id]")
except Exception:
first_row_element = None
self.driver.execute_script("arguments[0].click();", next_button)
time.sleep(2)
if first_row_element:
try:
self.wait.until(EC.staleness_of(first_row_element))
logger.info("Seitenwechsel erfolgreich verifiziert (alte Zeile ist verschwunden).")
except Exception as e:
logger.warning(f"Timeout beim Warten auf das Nachladen der Tabelle: {e}")
else:
time.sleep(2)
except Exception as e:
logger.error(f"Fehler beim Klicken auf den Weiter-Button: {e}")
break
return all_companies
def close(self): def close(self):
if self.driver: if self.driver: