Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
Floke
2025-07-08 10:21:13 +00:00
parent 8f4cb10301
commit d728c4c590
1 changed files with 58 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
dealfront_enrichment.py
View File

@@ -15,25 +15,29 @@ from config import Config, DEALFRONT_LOGIN_URL, DEALFRONT_CREDENTIALS_FILE, DEAL
OUTPUT_DIR = "/app/output"
# Logging-Konfiguration
LOG_LEVEL = logging.DEBUG if Config.DEBUG else logging.INFO
LOG_LEVEL = logging.INFO # Wir setzen den Standard auf INFO für eine saubere Ausgabe
LOG_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s'
# Root-Logger konfigurieren (gilt für alle Bibliotheken)
logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True)
# Root-Logger für die Konsolenausgabe konfigurieren
logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True, handlers=[logging.StreamHandler()])
# Logger für Selenium auf INFO setzen, um Spam zu vermeiden
logging.getLogger("selenium").setLevel(logging.INFO)
# Selenium-Logger auf WARNING setzen, um den Spam zu unterdrücken
logging.getLogger("selenium").setLevel(logging.WARNING)
# Logger für unser eigenes Skript
# Eigener Logger für unser Skript
logger = logging.getLogger(__name__)
# FileHandler hinzufügen, um in eine Datei zu loggen
log_filename = f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.log"
# FileHandler hinzufügen, um ALLES (inkl. DEBUG) in eine .txt-Datei zu schreiben
# Wichtig: Dieser Handler hat sein eigenes Level (DEBUG), um alles zu erfassen.
log_filename = f"dealfront_run_{time.strftime('%Y%m%d-%H%M%S')}.txt"
log_filepath = os.path.join(OUTPUT_DIR, log_filename)
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. Konsolenausgabe auf Level {logging.getLevelName(LOG_LEVEL)}. Log-Datei: {log_filepath}")
OUTPUT_DIR = "/app/output"
class DealfrontScraper:
@@ -108,77 +112,52 @@ class DealfrontScraper:
except Exception as e:
logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}")
def login_and_navigate_to_target(self):
def login_and_find_list(self, search_name):
"""
Führt den Login durch und klickt dann auf dem Dashboard auf den
"Quick Link", um zur Target-Seite zu gelangen.
Führt den gesamten Prozess vom Login bis zum Laden der Zielliste robust aus.
"""
if not self.driver: return False
username, password = self._load_credentials()
if not username or not password: return False
try:
# SCHRITT 1: LOGIN
# === 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(username)
self.driver.find_element(By.CSS_SELECTOR, "input[type='password']").send_keys(password)
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.")
# SCHRITT 2: AUF QUICK-LINK-KACHEL WARTEN UND KLICKEN
logger.info("Warte auf Dashboard und 'Prospects finden'-Link in den Quick Links.")
# Dieser XPath ist sehr spezifisch für den Link in der "Quick links"-Kachel
# === NAVIGATION ZUM TARGET BEREICH ===
logger.info("Warte auf Dashboard und den 'Prospects finden' Quick-Link...")
# Wir warten geduldig (bis zu 30s) auf die Kachel, die uns zur Target-Seite bringt
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()
logger.info("'Prospects finden' geklickt. Navigiere zur Target-Seite.")
# SCHRITT 3: NAVIGATION VERIFIZIEREN
logger.info("Warte auf die finale Target-Seite...")
verification_target_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']")
self.wait.until(EC.visibility_of_element_located(verification_target_selector))
# === 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()
logger.info("'Target'-Seite erfolgreich und vollständig geladen.")
# === 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) # Finale, großzügige Pause für das Laden der Tabellen-Daten via JS
logger.info("Zielseite mit Ergebnissen erfolgreich erreicht.")
return True
except Exception as e:
logger.critical(f"Der Prozess ist fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts() # Speichert Screenshot und HTML für die Analyse
return False
except Exception as e:
logger.critical(f"Login- oder Navigationsprozess fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts()
return False
def navigate_to_target(self):
"""
Navigiert zum 'Target'-Bereich und verifiziert den Erfolg in drei Schritten.
Dieser Ansatz ist maximal robust gegen Timing-Probleme von SPAs.
"""
try:
# SCHRITT 1: Befehl zur Navigation geben
logger.info(f"Gebe Navigationsbefehl zur Target-URL: {Config.DEALFRONT_TARGET_URL}")
self.driver.get(Config.DEALFRONT_TARGET_URL)
# SCHRITT 2: Warten, bis die URL in der Adresszeile sich tatsächlich ändert.
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}")
# SCHRITT 3: ERST JETZT auf ein sichtbares Element auf der neuen Seite warten.
verification_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']")
logger.info(f"Warte auf Sichtbarkeit des Verifizierungs-Elements auf der Target-Seite: {verification_selector}")
self.wait.until(EC.visibility_of_element_located(verification_selector))
logger.info("'Target'-Seite erfolgreich und vollständig geladen.")
return True
except Exception as e:
logger.critical(f"Navigation zur 'Target'-Seite endgültig 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:
@@ -204,36 +183,6 @@ class DealfrontScraper:
except Exception as e:
logger.warning(f"Fehler beim Schließen des Overlays, ignoriere und fahre fort: {e}")
self._save_debug_artifacts()
def load_search(self, search_name):
"""Lädt eine vordefinierte Suche anhand ihres Namens."""
try:
logger.info(f"Warte, bis die Liste der gespeicherten Suchen sichtbar ist...")
# Wir warten auf den Container, der alle Such-Items enthält.
# Ein guter Selektor ist das `ul`-Element mit der ID, die mit `eb-popup` beginnt.
# Oder allgemeiner: Ein Element, das den Text einer bekannten Suche enthält.
search_list_container_selector = (By.XPATH, f"//div[normalize-space()='{search_name}']")
self.wait.until(EC.visibility_of_element_located(search_list_container_selector))
logger.info(f"Liste gefunden. Klicke jetzt auf die vordefinierte Suche: '{search_name}'")
# Der XPath sucht nach einem Element (div, a, button), dessen Text exakt übereinstimmt.
search_link_selector = (By.XPATH, f"//*[normalize-space()='{search_name}']")
search_link = self.wait.until(EC.element_to_be_clickable(search_link_selector))
search_link.click()
# Verifizieren, dass die Ergebnisse geladen sind, indem wir auf die Ergebnistabelle warten.
results_table_header_selector = (By.XPATH, "//th[normalize-space()='Firma']")
self.wait.until(EC.visibility_of_element_located(results_table_header_selector))
logger.info(f"Suche '{search_name}' erfolgreich geladen und Ergebnisse angezeigt.")
time.sleep(3) # Feste Wartezeit, damit alle Daten nachgeladen sind.
return True
except Exception as e:
logger.critical(f"Laden der Suche '{search_name}' fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts()
return False
def extract_current_page_results(self):
"""
@@ -298,7 +247,7 @@ class DealfrontScraper:
self.driver.quit()
if __name__ == "__main__":
logger.info("Starte Dealfront Automatisierung - Finaler Workflow")
logger.info("Starte Dealfront Automatisierung - Finaler Durchbruchsversuch")
scraper = None
try:
@@ -306,35 +255,42 @@ if __name__ == "__main__":
if not scraper.driver:
raise Exception("WebDriver konnte nicht initialisiert werden.")
# === EINZIGER AUFRUF FÜR LOGIN UND NAVIGATION ===
if not scraper.login_and_navigate_to_target():
raise Exception("Login und Navigation fehlgeschlagen.")
# Ab hier geht es weiter wie geplant...
if not scraper.load_search(Config.TARGET_SEARCH_NAME):
raise Exception(f"Laden der Suche '{Config.TARGET_SEARCH_NAME}' fehlgeschlagen.")
scraper.handle_overlays()
# Ein einziger, robuster Aufruf für den gesamten Prozess bis zur Anzeige der Ergebnistabelle
if not scraper.login_and_find_list(Config.TARGET_SEARCH_NAME):
# Der Fehler wird bereits innerhalb der Methode geloggt und Artefakte werden gespeichert.
# Wir müssen hier nur noch den Prozess beenden.
raise Exception("Der Prozess vom Login bis zum Laden der Liste ist fehlgeschlagen. Details siehe Log.")
# Wenn wir hier ankommen, sind wir garantiert auf der richtigen Seite.
# Jetzt extrahieren wir die Daten.
companies = scraper.extract_current_page_results()
# Saubere Ausgabe der Ergebnisse in einer Tabelle
if companies:
df = pd.DataFrame(companies)
# Pandas-Optionen für eine vollständige, ungeschnittene Ausgabe
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("Keine Firmen auf der ersten Seite extrahiert oder gefunden.")
logger.info(f"{len(df)} Firmen erfolgreich extrahiert und in der Konsole ausgegeben.")
logger.info("Test erfolgreich abgeschlossen. Warte vor dem Schließen...")
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:
# Hier fangen wir nur noch die Exceptions aus dem __main__-Block selbst
logger.critical(f"Ein kritischer Fehler ist im Hauptprozess aufgetreten: {e}", exc_info=False)
finally:
if scraper: