import os import json import time import logging import pandas as pd from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.chrome.options import Options as ChromeOptions from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.common.exceptions import TimeoutException, NoSuchElementException # --- Konfiguration --- class Config: LOGIN_URL = "https://app.dealfront.com/login" TARGET_URL = "https://app.dealfront.com/t/prospector/companies" SEARCH_NAME = "Facility Management" # <-- PASSEN SIE DIES AN IHRE GESPEICHERTE SUCHE AN CREDENTIALS_FILE = "/app/dealfront_credentials.json" OUTPUT_DIR = "/app/output" # --- Logging Setup --- LOG_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s' logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, force=True) 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") file_handler = logging.FileHandler(log_filepath, mode='w', encoding='utf-8') file_handler.setFormatter(logging.Formatter(LOG_FORMAT)) logging.getLogger().addHandler(file_handler) class DealfrontScraper: def __init__(self): logger.info("Initialisiere WebDriver...") chrome_options = ChromeOptions() chrome_options.add_experimental_option("prefs", {"profile.managed_default_content_settings.images": 2}) chrome_options.add_argument("--headless=new") chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--window-size=1920,1200") try: self.driver = webdriver.Chrome(options=chrome_options) except Exception as e: logger.critical("WebDriver konnte nicht initialisiert werden.", exc_info=True) raise self.wait = WebDriverWait(self.driver, 30) self.username, self.password = self._load_credentials() if not self.username or not self.password: raise ValueError("Credentials konnten nicht geladen werden. Breche ab.") logger.info("WebDriver erfolgreich initialisiert.") def _load_credentials(self): try: with open(Config.CREDENTIALS_FILE, 'r', encoding='utf-8') as f: creds = json.load(f) return creds.get("username"), creds.get("password") except Exception: logger.error(f"Credentials-Datei {Config.CREDENTIALS_FILE} nicht gefunden.") return None, None def _save_debug_artifacts(self, suffix=""): try: timestamp = time.strftime("%Y%m%d-%H%M%S") filename_base = os.path.join(Config.OUTPUT_DIR, f"error_{suffix}_{timestamp}") self.driver.save_screenshot(f"{filename_base}.png") with open(f"{filename_base}.html", "w", encoding="utf-8") as f: f.write(self.driver.page_source) logger.error(f"Debug-Artefakte gespeichert: {filename_base}.*") except Exception as e: logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}") def run(self): # 1. LOGIN logger.info(f"Navigiere zu: {Config.LOGIN_URL}") self.driver.get(Config.LOGIN_URL) 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.XPATH, "//button[normalize-space()='Log in']").click() logger.info("Login-Befehl gesendet. Warte 5s auf Session-Etablierung.") time.sleep(5) # 2. NAVIGATION & SUCHE LADEN logger.info(f"Navigiere direkt zur Target-Seite und lade Suche: '{Config.SEARCH_NAME}'") self.driver.get(Config.TARGET_URL) self.wait.until(EC.element_to_be_clickable((By.XPATH, f"//*[normalize-space()='{Config.SEARCH_NAME}']"))).click() # 3. ERGEBNISSE EXTRAHIEREN logger.info("Suche geladen. Extrahiere Ergebnisse der ersten Seite.") self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "table#t-result-table"))) time.sleep(3) # Letzte kurze Pause für das Rendering 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") logger.info(f"{len(company_elements)} Firmen und {len(website_elements)} Webseiten gefunden.") results = [] for i in range(len(company_elements)): name = company_elements[i].get_attribute("title").strip() website = website_elements[i].text.strip() if i < len(website_elements) else "N/A" results.append({'name': name, 'website': website}) return results def scrape_all_pages(self): """ Iteriert durch alle Ergebnisseiten, extrahiert die Daten und gibt eine Gesamtliste zurück. """ all_companies = [] page_number = 1 while True: logger.info(f"--- Verarbeite Seite {page_number} ---") # Warten, bis die Tabelle sichtbar ist table_selector = (By.CSS_SELECTOR, "table#t-result-table") self.wait.until(EC.visibility_of_element_located(table_selector)) # ID des ersten Eintrags merken, um Seitenwechsel zu verifizieren try: first_row_id = self.driver.find_element(By.CSS_SELECTOR, "table#t-result-table tbody tr").get_attribute("id") except NoSuchElementException: logger.warning("Konnte keine Datenzeilen auf der Seite finden. Beende Paginierung.") break # Daten von der aktuellen Seite extrahieren page_results = self.extract_current_page_results() if not page_results: logger.info("Keine weiteren Ergebnisse auf der Seite gefunden. Paginierung abgeschlossen.") break all_companies.extend(page_results) logger.info(f"Bisher {len(all_companies)} Firmen insgesamt gefunden.") # "Weiter"-Button finden und prüfen, ob er deaktiviert ist try: next_button_selector = (By.XPATH, "//a[@class='eb-pagination-button'][.//svg[contains(@class, 'fa-angle-right')]]") next_button = self.driver.find_element(*next_button_selector) if "disabled" in next_button.get_attribute("class"): logger.info("'Weiter'-Button ist deaktiviert. Letzte Seite erreicht.") break # Auf den "Weiter"-Button klicken next_button.click() logger.info(f"Auf Seite {page_number + 1} geblättert. Warte auf neue Inhalte...") page_number += 1 # Warten, bis die alte erste Zeile verschwunden ist (Staleness) # Dies ist der robusteste Weg, um zu bestätigen, dass die Seite neu geladen wurde. old_first_row = self.driver.find_element(By.ID, first_row_id) self.wait.until(EC.staleness_of(old_first_row)) except NoSuchElementException: logger.info("Kein 'Weiter'-Button mehr gefunden. Paginierung abgeschlossen.") break return all_companies def close(self): if self.driver: self.driver.quit() if __name__ == "__main__": logger.info("Starte Dealfront Automatisierung - Paginierungstest") scraper = None try: scraper = DealfrontScraper() if not scraper.login(): raise Exception("Login-Phase fehlgeschlagen") if not scraper.navigate_and_load_search(Config.SEARCH_NAME): raise Exception("Navigation und Laden der Suche fehlgeschlagen.") # Ruft die neue Methode auf, die durch alle Seiten blättert all_companies = scraper.scrape_all_pages() if all_companies: df = pd.DataFrame(all_companies) logger.info(f"===== Insgesamt {len(df)} Firmen auf allen Seiten extrahiert =====") # Optionale Ausgabe der kompletten Liste in der Konsole # pd.set_option('display.max_rows', None) # pd.set_option('display.width', 120) # print(df.to_string(index=False)) # Speichere die Ergebnisse in einer CSV-Datei im Output-Ordner output_csv_path = os.path.join(Config.OUTPUT_DIR, f"dealfront_results_{time.strftime('%Y%m%d-%H%M%S')}.csv") df.to_csv(output_csv_path, index=False, sep=';', encoding='utf-8-sig') logger.info(f"Ergebnisse erfolgreich in '{output_csv_path}' gespeichert.") else: logger.warning("Keine Firmen konnten extrahiert werden.") except Exception as e: logger.critical(f"Ein kritischer Fehler ist im Hauptprozess aufgetreten: {e}", exc_info=True) finally: if scraper: scraper.close() logger.info("Dealfront Automatisierung beendet.")``` ### Zusammenfassung der Änderungen * **Paginierungslogik:** Die neue Methode `scrape_all_pages` kümmert sich um das Durchblättern der Seiten. * **Stabiles Warten:** Anstatt fester `time.sleep`-Pausen warten wir nun mit `EC.staleness_of`, bis die alten Daten verschwunden sind. Das ist schnell und zuverlässig. * **CSV-Export:** Die Ausgabe in der Konsole kann sehr lang werden. Daher werden die Ergebnisse jetzt direkt in eine **CSV-Datei** im `output`-Ordner gespeichert. Das ist sauberer und besser für die Weiterverarbeitung. ### Nächster Schritt 1. **Code ändern:** Fügen Sie die neue Methode hinzu und ersetzen Sie den `__main__`-Block. 2. **Testlauf in der Container-Shell:** ```bash python3 dealfront_enrichment.py ``` **Erwartetes Ergebnis:** Das Skript wird Seite für Seite durchgehen und die Logs entsprechend ausgeben. Am Ende finden Sie eine neue CSV-Datei in Ihrem `output`-Ordner, die alle extrahierten Firmennamen und Webseiten enthält.