dealfront_enrichment.py aktualisiert

This commit is contained in:
2025-07-03 20:19:42 +00:00
parent fd54bc8e62
commit 2c56828f76

View File

@@ -81,8 +81,8 @@ class DealfrontScraper:
def login_and_navigate_to_target(self):
"""
Führt Login durch und navigiert zur Target-Seite via Klick auf den "Quick Link".
Dieser Ansatz ist maximal robust.
Führt den Login durch und navigiert zur Target-URL.
Dieser kombinierte Ansatz wartet nach jeder Aktion explizit auf den Erfolg.
"""
if not self.driver:
return False
@@ -92,38 +92,37 @@ class DealfrontScraper:
return False
try:
# --- SCHRITT 1: LOGIN ---
# --- SCHRITT 1: LOGIN-SEITE LADEN ---
logger.info(f"Navigiere zur Login-Seite: {DEALFRONT_LOGIN_URL}")
self.driver.get(DEALFRONT_LOGIN_URL)
# --- SCHRITT 2: DATEN EINGEBEN UND KLICKEN ---
logger.info("Warte auf Login-Formular und fülle es aus...")
email_field = self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "input[name='email']")))
email_field.send_keys(username)
password_field = self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "input[type='password']")))
password_field = self.driver.find_element(By.CSS_SELECTOR, "input[type='password']")
password_field.send_keys(password)
login_button = self.wait.until(EC.element_to_be_clickable((By.XPATH, "//button[normalize-space()='Log in']")))
login_button = self.driver.find_element(By.XPATH, "//button[normalize-space()='Log in']")
login_button.click()
# --- SCHRITT 2: NAVIGATION VIA QUICK-LINK-KACHEL ---
logger.info("Login-Befehl gesendet. Warte auf Dashboard und 'Prospects finden'-Link.")
# Dieser XPath zielt auf den Link in der "Quick links"-Kachel
prospects_link_selector = (By.XPATH, "//a[@data-test-target-product-tile]")
prospects_link = self.wait.until(EC.element_to_be_clickable(prospects_link_selector))
logger.info("'Prospects finden'-Link gefunden. Klicke darauf...")
prospects_link.click()
# --- SCHRITT 3: NAVIGATION VERIFIZIEREN ---
url_part_to_wait_for = "/t/prospector/"
logger.info(f"Warte, bis die Browser-URL '{url_part_to_wait_for}' enthält...")
self.wait.until(EC.url_contains(url_part_to_wait_for))
logger.info(f"URL-Wechsel bestätigt. Aktuelle URL: {self.driver.current_url}")
logger.info("Login-Button geklickt.")
# --- SCHRITT 3: AUF ERFOLGREICHEN LOGIN WARTEN (Dashboard) ---
logger.info("Warte auf Dashboard nach dem Login...")
verification_dashboard_selector = (By.XPATH, "//input[@data-cy='header-search-input']")
self.wait.until(EC.visibility_of_element_located(verification_dashboard_selector))
logger.info("Dashboard erfolgreich erreicht.")
time.sleep(2) # Eine kleine Pause, damit alle Hintergrund-Skripte laden können.
# --- SCHRITT 4: DIREKTE NAVIGATION ZUR TARGET-URL ---
logger.info(f"Navigiere direkt zur Target-URL: {Config.DEALFRONT_TARGET_URL}")
self.driver.get(Config.DEALFRONT_TARGET_URL)
# --- SCHRITT 5: ERFOLG DER NAVIGATION VERIFIZIEREN ---
logger.info("Warte auf Bestätigung der Target-Seite...")
verification_target_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']")
self.wait.until(EC.visibility_of_element_located(verification_target_selector))
logger.info("'Target'-Seite erfolgreich und vollständig geladen.")
return True
except Exception as e:
@@ -221,7 +220,7 @@ if __name__ == "__main__":
if not scraper.driver:
raise Exception("WebDriver konnte nicht initialisiert werden.")
# Führe den kombinierten Login- und Navigationsschritt aus
# === NEUER AUFRUF DER KOMBINIERTEN FUNKTION ===
if not scraper.login_and_navigate_to_target():
raise Exception("Login und Navigation fehlgeschlagen.")
@@ -229,31 +228,19 @@ if __name__ == "__main__":
if not scraper.load_search(Config.TARGET_SEARCH_NAME):
raise Exception(f"Laden der Suche '{Config.TARGET_SEARCH_NAME}' fehlgeschlagen.")
# Ergebnisse extrahieren
# Ergebnisse extrahieren und ausgeben
companies = scraper.extract_current_page_results()
# === NEUE, SAUBERE AUSGABE ===
if companies:
# Erstelle einen pandas DataFrame aus der Ergebnisliste
df = pd.DataFrame(companies)
# Konfiguriere pandas, um den vollen Text in den Spalten anzuzeigen
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)
if not df.empty:
print(df.to_string(index=False))
else:
print(" DataFrame ist leer, obwohl Ergebnisse vorhanden waren. Überprüfung nötig. ".center(80, "-"))
print(df.to_string(index=False))
print("="*80 + "\n")
logger.info(f"{len(df)} Firmen erfolgreich in der Konsole ausgegeben.")
else:
logger.warning("Keine Firmen auf der ersten Seite extrahiert oder gefunden.")