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.service import Service 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 # Importiere Konfigurationen from config import Config, DEALFRONT_LOGIN_URL, DEALFRONT_CREDENTIALS_FILE, DEALFRONT_TARGET_URL, TARGET_SEARCH_NAME # Definiere einen festen Ausgabeordner, der im Dockerfile erstellt wird OUTPUT_DIR = "/app/output" # Logging-Konfiguration LOG_LEVEL = logging.INFO LOG_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s' logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True, handlers=[logging.StreamHandler()]) logging.getLogger("selenium").setLevel(logging.WARNING) # Selenium-Spam reduzieren logger = logging.getLogger(__name__) # FileHandler hinzufügen, um in eine .txt-Datei zu loggen log_filename = f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.txt" log_filepath = os.path.join(OUTPUT_DIR, log_filename) try: file_handler = logging.FileHandler(log_filepath, mode='w', encoding='utf-8') file_handler.setLevel(logging.DEBUG) # Alles in die Datei schreiben file_handler.setFormatter(logging.Formatter(LOG_FORMAT)) logging.getLogger().addHandler(file_handler) logger.info(f"Logging konfiguriert. Konsole auf Level {logging.getLevelName(LOG_LEVEL)}. Log-Datei: {log_filepath}") except FileNotFoundError: logger.error(f"Konnte Log-Datei nicht erstellen. Das Verzeichnis '{OUTPUT_DIR}' existiert möglicherweise nicht im Container.") logger.error("Stellen Sie sicher, dass das Volume-Mapping korrekt ist: -v \"$(pwd)/output:/app/output\"") class DealfrontScraper: """ Kapselt alle Interaktionen mit der Dealfront-Plattform mittels Selenium. """ def __init__(self): logger.info("Initialisiere den DealfrontScraper...") chrome_options = ChromeOptions() prefs = {"profile.managed_default_content_settings.images": 2} chrome_options.add_experimental_option("prefs", prefs) chrome_options.add_argument("--headless") chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--window-size=1920,1080") # Service-Objekt, das explizit den system-installierten Treiber verwendet service = Service(executable_path='/usr/bin/chromedriver') try: self.driver = webdriver.Chrome(service=service, options=chrome_options) except Exception as e: logger.critical(f"WebDriver konnte nicht initialisiert werden. Fehler: {e}", exc_info=True) self.driver = None raise self.wait = WebDriverWait(self.driver, 30) self.username, self.password = self._load_credentials() logger.info("WebDriver erfolgreich initialisiert.") def _load_credentials(self): try: with open(DEALFRONT_CREDENTIALS_FILE, 'r') as f: creds = json.load(f) username = creds.get("username") password = creds.get("password") if not username or "DEIN_DEALFRONT_BENUTZERNAME" in username or not password: logger.error(f"Zugangsdaten in '{DEALFRONT_CREDENTIALS_FILE}' sind ungültig.") return None, None return username, password except FileNotFoundError: logger.error(f"Credentials-Datei nicht gefunden: '{DEALFRONT_CREDENTIALS_FILE}'") return None, None return None, None def _save_debug_artifacts(self): try: os.makedirs(OUTPUT_DIR, exist_ok=True) timestamp = time.strftime("%Y%m%d-%H%M%S") screenshot_filepath = os.path.join(OUTPUT_DIR, f"error_{timestamp}.png") html_filepath = os.path.join(OUTPUT_DIR, f"error_{timestamp}.html") self.driver.save_screenshot(screenshot_filepath) logger.error(f"Screenshot '{screenshot_filepath}' wurde für die Analyse gespeichert.") with open(html_filepath, "w", encoding="utf-8") as f: f.write(self.driver.page_source) logger.error(f"HTML-Quellcode '{html_filepath}' wurde für die Analyse gespeichert.") except Exception as e: logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}") def login_and_find_list(self, search_name): """Führt den gesamten Prozess vom Login bis zum Laden der Zielliste robust aus.""" try: # === LOGIN === logger.info(f"Navigiere zur Login-Seite: {DEALFRONT_LOGIN_URL}") self.driver.get(DEALFRONT_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.") # === NAVIGATION ZUM TARGET BEREICH === 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.") # === LADEN DER SPEZIFISCHEN SUCHE === 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 = self.wait.until(EC.element_to_be_clickable(search_item_selector)) search_item.click() # === VERIFIZIERUNG UND WARTEN AUF TABELLENDATEN === logger.info(f"Suche '{search_name}' geladen. Warte auf das Rendern der Ergebnistabelle.") table_header_selector = (By.XPATH, "//th[normalize-space()='Firma']") self.wait.until(EC.visibility_of_element_located(table_header_selector)) time.sleep(5) logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.") return True except Exception as e: logger.critical(f"Der Prozess vom Login bis zum Finden der Liste ist fehlgeschlagen: {type(e).__name__}", exc_info=True) self._save_debug_artifacts() return False def handle_overlays(self): """Sucht nach bekannten Popups/Overlays und schließt sie.""" try: short_wait = WebDriverWait(self.driver, 5) close_button_xpath = "//button[@aria-label='Schließen' or @aria-label='Close']" logger.info("Suche nach bekannten Overlays/Popups...") close_button = short_wait.until(EC.element_to_be_clickable((By.XPATH, close_button_xpath))) logger.info("Schließen-Button für Overlay gefunden. Klicke darauf.") close_button.click() time.sleep(1) except TimeoutException: logger.info("Kein Overlay/Popup gefunden. Fahre fort.") except Exception as e: logger.warning(f"Fehler beim Schließen des Overlays, ignoriere und fahre fort: {e}") def extract_current_page_results(self): """Extrahiert die Firmennamen und Webseiten.""" try: logger.info("Extrahiere Ergebnisse von der aktuellen Seite...") results = [] rows_selector = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id]") self.wait.until(EC.presence_of_element_located(rows_selector)) time.sleep(3) rows = self.driver.find_elements(*rows_selector) if not rows: logger.warning("Keine Ergebniszeilen (tr[id]) gefunden.") return [] logger.info(f"{len(rows)} Firmen-Datenzeilen zur Verarbeitung gefunden.") for i, row in enumerate(rows, 1): try: company_name = row.find_element(By.CSS_SELECTOR, ".sticky-column a.t-highlight-text").get_attribute("title").strip() website = row.find_element(By.CSS_SELECTOR, "a.text-gray-400.t-highlight-text").text.strip() results.append({'name': company_name, 'website': website}) except NoSuchElementException: logger.warning(f"Zeile {i}: Name oder Webseite nicht extrahierbar. Überspringe.") continue logger.info(f"Extraktion abgeschlossen. {len(results)} Firmen gefunden.") return results except Exception as e: logger.error(f"Schwerwiegender Fehler bei der Extraktion: {type(e).__name__}", exc_info=True) self._save_debug_artifacts() return [] def close(self): if self.driver: logger.info("Schließe den WebDriver.") self.driver.quit() if __name__ == "__main__": logger.info("Starte Dealfront Automatisierung - Finaler Durchbruchsversuch") scraper = None try: scraper = DealfrontScraper() if not scraper.driver: raise Exception("WebDriver konnte nicht initialisiert werden.") if not scraper.login_and_find_list(Config.TARGET_SEARCH_NAME): raise Exception("Der Prozess vom Login bis zum Laden der Liste ist fehlgeschlagen. Details siehe Log.") scraper.handle_overlays() companies = scraper.extract_current_page_results() if companies: df = pd.DataFrame(companies) pd.set_option('display.max_rows', None) pd.set_option('display.max_columns', None) pd.set_option('display.width', 1000) pd.set_option('display.max_colwidth', None) print("\n" + "="*80) print(" EXTRAHIERTE FIRMEN (ERSTE SEITE) ".center(80, "=")) print("="*80) print(df.to_string(index=False)) print("="*80 + "\n") else: logger.warning("Obwohl die Seite geladen wurde, konnten keine Firmen extrahiert werden. Bitte HTML-Dump prüfen.") logger.info("Phase 2a Test erfolgreich abgeschlossen. Warte vor dem Schließen...") time.sleep(10) except Exception as e: logger.critical(f"Ein kritischer Fehler ist im Hauptprozess aufgetreten: {e}", exc_info=False) finally: if scraper: scraper.close() logger.info("Dealfront Automatisierung beendet.")