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:52:35 +00:00
parent 97f4709d51
commit 99fc7db8fa
1 changed files with 31 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
dealfront_enrichment.py
View File

@@ -25,7 +25,7 @@ logging.getLogger("selenium.webdriver.remote").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
os.makedirs(Config.OUTPUT_DIR, exist_ok=True)
log_filepath = os.path.join(Config.OUTPUT_DIR, f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.log")
log_filepath = os.path.join(Config.OUTPUT_DIR, f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.txt")
file_handler = logging.FileHandler(log_filepath, mode='w', encoding='utf-8')
file_handler.setFormatter(logging.Formatter(LOG_FORMAT))
logging.getLogger().addHandler(file_handler)
@@ -131,68 +131,58 @@ class DealfrontScraper:
return results
def scrape_all_pages(self, max_pages=6):
def scrape_all_pages(self, max_pages=10):
"""
Iteriert durch alle Ergebnisseiten, scrollt auf jeder Seite nach unten,
um alle Daten zu laden, und extrahiert sie dann. Verhindert Duplikate.
Iteriert durch alle Ergebnisseiten, scrollt auf jeder Seite vollständig,
extrahiert die Daten und klickt auf den korrekten "Weiter"-Button.
"""
all_companies = {} # Verwende ein Dictionary, um Duplikate zu vermeiden
all_companies = []
for page_number in range(1, max_pages + 1):
logger.info(f"--- Verarbeite Seite {page_number} ---")
# 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))
self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "table#t-result-table")))
# --- ROBUSTES SCROLLEN ---
logger.info("Scrolle Seite nach unten, um alle Zeilen zu laden...")
scroll_container = self.driver.find_element(By.CSS_SELECTOR, ".scroll-viewport")
last_height = self.driver.execute_script("return arguments[0].scrollHeight", scroll_container)
# --- 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")
self.driver.execute_script("arguments[0].scrollTo(0, arguments[0].scrollHeight);", scroll_container)
time.sleep(2.5) # Wichtige Pause, damit neue Inhalte nachladen können
new_height = self.driver.execute_script("return arguments[0].scrollHeight", scroll_container)
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()
if not page_results:
logger.warning("Konnte auf dieser Seite keine Ergebnisse extrahieren, obwohl gescrollt wurde.")
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
logger.info(f"Seite {page_number}: {len(page_results)} Firmen gefunden. Gesamt einzigartig: {len(all_companies)}")
all_companies.extend(page_results)
logger.info(f"Seite {page_number}: {len(page_results)} Firmen gefunden. Gesamt: {len(all_companies)}")
# --- Paginierungs-Logik ---
# --- FINALE PAGINIERUNGS-LOGIK ---
try:
# Wir merken uns die ID der LETZTEN Zeile, um den Wechsel zu verifizieren
last_row_id = self.driver.find_element(By.XPATH, "(//table[@id='t-result-table']/tbody/tr[@id])[last()]").get_attribute("id")
next_button_selector = (By.XPATH, "//nav[contains(@class, 'eb-pagination')]//a[not(contains(@class, 'disabled'))][last()]")
# Eindeutiger XPath für den "Weiter"-Pfeil-Button, der nicht deaktiviert ist
next_button_selector = (By.XPATH, "//a[contains(@class, 'eb-pagination-button') and not(contains(@class, 'disabled')) and .//path[starts-with(@d, 'M8.293')]]")
next_button = self.driver.find_element(*next_button_selector)
if "fa-angle-right" not in next_button.get_attribute('innerHTML'):
logger.info("Letzte Seite erreicht (kein 'Weiter'-Pfeil im letzten Button).")
break
# Den Button in den sichtbaren Bereich scrollen, falls er verdeckt ist
self.driver.execute_script("arguments[0].scrollIntoView(true);", next_button)
time.sleep(0.5)
next_button.click()
logger.info(f"Auf Seite {page_number + 1} geblättert...")
# Warten auf das Neuladen der Tabelle
self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "table#t-result-table")))
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.")
except NoSuchElementException:
logger.info("Kein 'Weiter'-Button mehr gefunden. Paginierung abgeschlossen.")
break
return list(all_companies.values())
return all_companies
def close(self):
if self.driver: