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-03 12:24:02 +00:00
parent 5d83a1df5d
commit 5ddd7c05db
1 changed files with 33 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
dealfront_enrichment.py
View File

@@ -8,22 +8,21 @@ from selenium.webdriver.chrome.options import Options as ChromeOptions
from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, NoSuchElementException from selenium.common.exceptions import TimeoutException, NoSuchElementException
from selenium.webdriver.common.keys import Keys
# Wichtig: Die zentralen Imports zuerst from config import Config, DEALFRONT_LOGIN_URL, DEALFRONT_CREDENTIALS_FILE, DEALFRONT_TARGET_URL, TARGET_SEARCH_NAME
from config import Config, DEALFRONT_LOGIN_URL, DEALFRONT_CREDENTIALS_FILE, DEALFRONT_TARGET_URL
# Logging-Konfiguration, eigenständig für dieses Skript # Logging-Konfiguration
LOG_LEVEL = logging.DEBUG if Config.DEBUG else logging.INFO LOG_LEVEL = logging.DEBUG if Config.DEBUG else logging.INFO
LOG_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s' LOG_FORMAT = '%(asctime)s - %(levelname)-8s - %(name)-25s - %(message)s'
logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True) logging.basicConfig(level=LOG_LEVEL, format=LOG_FORMAT, force=True)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Definiere einen festen Ausgabeordner, der im Dockerfile erstellt wird
OUTPUT_DIR = "/app/output" OUTPUT_DIR = "/app/output"
class DealfrontScraper: class DealfrontScraper:
def __init__(self): def __init__(self):
logger.info("Initialisiere den DealfrontScraper und den Chrome WebDriver.") logger.info("Initialisiere DealfrontScraper und Chrome WebDriver.")
chrome_options = ChromeOptions() chrome_options = ChromeOptions()
chrome_options.add_argument("--headless") chrome_options.add_argument("--headless")
chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--no-sandbox")
@@ -41,10 +40,11 @@ class DealfrontScraper:
self.driver = None self.driver = None
raise raise
self.wait = WebDriverWait(self.driver, 20) self.wait = WebDriverWait(self.driver, 30) # Timeout auf 30s erhöht für mehr Stabilität
logger.info("WebDriver erfolgreich initialisiert.") logger.info("WebDriver erfolgreich initialisiert.")
def _load_credentials(self): def _load_credentials(self):
# (Diese Methode bleibt unverändert)
try: try:
with open(DEALFRONT_CREDENTIALS_FILE, 'r') as f: with open(DEALFRONT_CREDENTIALS_FILE, 'r') as f:
creds = json.load(f) creds = json.load(f)
@@ -62,29 +62,26 @@ class DealfrontScraper:
return None, None return None, None
def _save_debug_artifacts(self): def _save_debug_artifacts(self):
"""Speichert einen Screenshot UND den HTML-Quellcode im Fehlerfall.""" # (Diese Methode bleibt unverändert, aber mit neuem Namen)
try: try:
os.makedirs(OUTPUT_DIR, exist_ok=True) os.makedirs(OUTPUT_DIR, exist_ok=True)
timestamp = time.strftime("%Y%m%d-%H%M%S") timestamp = time.strftime("%Y%m%d-%H%M%S")
# 1. Screenshot speichern
screenshot_filepath = os.path.join(OUTPUT_DIR, f"error_{timestamp}.png") screenshot_filepath = os.path.join(OUTPUT_DIR, f"error_{timestamp}.png")
self.driver.save_screenshot(screenshot_filepath) self.driver.save_screenshot(screenshot_filepath)
logger.error(f"Screenshot '{screenshot_filepath}' wurde für die Analyse gespeichert.") logger.error(f"Screenshot '{screenshot_filepath}' wurde für die Analyse gespeichert.")
# 2. HTML-Quellcode speichern
html_filepath = os.path.join(OUTPUT_DIR, f"error_{timestamp}.html") html_filepath = os.path.join(OUTPUT_DIR, f"error_{timestamp}.html")
with open(html_filepath, "w", encoding="utf-8") as f: with open(html_filepath, "w", encoding="utf-8") as f:
f.write(self.driver.page_source) f.write(self.driver.page_source)
logger.error(f"HTML-Quellcode '{html_filepath}' wurde für die Analyse gespeichert.") logger.error(f"HTML-Quellcode '{html_filepath}' wurde für die Analyse gespeichert.")
except Exception as e: except Exception as e:
logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}") logger.error(f"Konnte Debug-Artefakte nicht speichern: {e}")
def login_and_navigate_to_target(self): def login_and_navigate_to_target(self):
""" """
Führt den Login durch UND navigiert direkt zur Target-Seite. Führt Login durch und navigiert zur Target-Seite via Klick auf den "Quick Link".
Gibt nur True zurück, wenn BEIDE Schritte erfolgreich sind. Dieser Ansatz ist maximal robust.
""" """
if not self.driver: if not self.driver:
return False return False
@@ -98,36 +95,34 @@ class DealfrontScraper:
logger.info(f"Navigiere zur Login-Seite: {DEALFRONT_LOGIN_URL}") logger.info(f"Navigiere zur Login-Seite: {DEALFRONT_LOGIN_URL}")
self.driver.get(DEALFRONT_LOGIN_URL) self.driver.get(DEALFRONT_LOGIN_URL)
email_selector = (By.CSS_SELECTOR, "input[name='email']") email_field = self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "input[name='email']")))
email_field = self.wait.until(EC.visibility_of_element_located(email_selector))
email_field.send_keys(username) email_field.send_keys(username)
password_selector = (By.CSS_SELECTOR, "input[type='password']") password_field = self.wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "input[type='password']")))
password_field = self.wait.until(EC.visibility_of_element_located(password_selector))
password_field.send_keys(password) password_field.send_keys(password)
login_button_selector = (By.XPATH, "//button[normalize-space()='Log in']") login_button = self.wait.until(EC.element_to_be_clickable((By.XPATH, "//button[normalize-space()='Log in']")))
login_button = self.wait.until(EC.element_to_be_clickable(login_button_selector))
login_button.click() login_button.click()
# Login Verifizierung # --- SCHRITT 2: NAVIGATION VIA QUICK-LINK-KACHEL ---
verification_login_selector = (By.XPATH, "//input[@data-cy='header-search-input']") logger.info("Login-Befehl gesendet. Warte auf Dashboard und 'Prospects finden'-Link.")
self.wait.until(EC.visibility_of_element_located(verification_login_selector))
logger.info("Login auf Dashboard erfolgreich verifiziert.") # Dieser XPath zielt auf den Link in der "Quick links"-Kachel
prospects_link_selector = (By.XPATH, "//a[@data-test-target-product-tile]")
# --- SCHRITT 2: DIREKTE NAVIGATION ZUR TARGET-SEITE --- prospects_link = self.wait.until(EC.element_to_be_clickable(prospects_link_selector))
logger.info(f"Navigiere jetzt direkt zur Target-URL: {Config.DEALFRONT_TARGET_URL}") logger.info("'Prospects finden'-Link gefunden. Klicke darauf...")
self.driver.get(Config.DEALFRONT_TARGET_URL) prospects_link.click()
# Navigation Verifizierung # --- SCHRITT 3: NAVIGATION VERIFIZIEREN ---
url_part_to_wait_for = "/t/prospector/" 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)) self.wait.until(EC.url_contains(url_part_to_wait_for))
logger.info(f"URL-Wechsel zu Target bestätigt. Aktuelle URL: {self.driver.current_url}") logger.info(f"URL-Wechsel bestätigt. Aktuelle URL: {self.driver.current_url}")
verification_target_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']") verification_target_selector = (By.XPATH, "//button[normalize-space()='+ Neue Suche']")
self.wait.until(EC.visibility_of_element_located(verification_target_selector)) self.wait.until(EC.visibility_of_element_located(verification_target_selector))
logger.info("'Target'-Seite erfolgreich und vollständig geladen.")
logger.info("'Target'-Seite erfolgreich und vollständig geladen.")
return True return True
except Exception as e: except Exception as e:
@@ -135,92 +130,47 @@ class DealfrontScraper:
self._save_debug_artifacts() self._save_debug_artifacts()
return False 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.
# Das ist der kritischste Schritt, um zu bestätigen, dass der Navigations-Request
# vom Browser verarbeitet wurde. Wir suchen nach dem Teil "/t/prospector/".
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.
# Dies bestätigt, dass die Seite nicht nur navigiert, sondern auch gerendert wurde.
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 load_search(self, search_name): def load_search(self, search_name):
"""Lädt eine vordefinierte Suche anhand ihres Namens.""" # (Diese Methode bleibt vorerst unverändert)
try: try:
logger.info(f"Suche und lade die vordefinierte Suche: '{search_name}'") logger.info(f"Suche und lade die vordefinierte Suche: '{search_name}'")
# Der XPath sucht nach einem Link, dessen Text den Suchnamen enthält search_link_selector = (By.XPATH, f"//div[contains(@class, 'truncate') and normalize-space()='{search_name}']")
search_link_selector = (By.XPATH, f"//a[contains(., '{search_name}')] | //div[contains(., '{search_name}') and contains(@class, 'list-item')]")
search_link = self.wait.until(EC.element_to_be_clickable(search_link_selector)) search_link = self.wait.until(EC.element_to_be_clickable(search_link_selector))
search_link.click() search_link.click()
# Verifizieren, dass die Ergebnisse geladen sind, indem wir auf die Ergebnistabelle warten.
# Ein guter Indikator ist die Tabellenüberschrift "Firma".
results_table_header_selector = (By.XPATH, "//th[normalize-space()='Firma']") results_table_header_selector = (By.XPATH, "//th[normalize-space()='Firma']")
self.wait.until(EC.visibility_of_element_located(results_table_header_selector)) self.wait.until(EC.visibility_of_element_located(results_table_header_selector))
logger.info(f"Suche '{search_name}' erfolgreich geladen und Ergebnisse angezeigt.") logger.info(f"Suche '{search_name}' erfolgreich geladen und Ergebnisse angezeigt.")
# Eine kleine, feste Wartezeit kann helfen, damit alle Daten nachgeladen sind.
time.sleep(3) time.sleep(3)
return True return True
except Exception as e: except Exception as e:
logger.critical(f"Laden der Suche '{search_name}' fehlgeschlagen: {type(e).__name__}", exc_info=True) logger.critical(f"Laden der Suche '{search_name}' fehlgeschlagen: {type(e).__name__}", exc_info=True)
self._save_debug_artifacts() self._save_debug_artifacts()
return False return False
def extract_current_page_results(self): def extract_current_page_results(self):
"""Extrahiert die Firmennamen und Webseiten von der aktuellen Ergebnisseite.""" # (Diese Methode bleibt vorerst unverändert)
logger.info("Extrahiere Ergebnisse von der aktuellen Seite...")
results = []
try: try:
logger.info("Extrahiere Ergebnisse von der aktuellen Seite...")
results = []
# Finde alle Zeilen in der Ergebnistabelle. Wir gehen davon aus, dass es `tr`-Elemente sind.
# Ein stabiler Selektor könnte die `data-test`-Attribute nutzen, falls vorhanden.
rows_selector = (By.XPATH, "//tbody/tr") rows_selector = (By.XPATH, "//tbody/tr")
rows = self.driver.find_elements(*rows_selector) rows = self.driver.find_elements(*rows_selector)
if not rows: if not rows:
logger.warning("Keine Ergebniszeilen auf der Seite gefunden.") logger.warning("Keine Ergebniszeilen auf der Seite gefunden.")
self._save_debug_artifacts()
return [] return []
logger.info(f"{len(rows)} Ergebniszeilen gefunden.") logger.info(f"{len(rows)} Ergebniszeilen gefunden.")
for row in rows: for row in rows:
try: try:
# Extrahiere Firmenname und Webseite.
# Diese XPaths sind Annahmen und müssen ggf. nach Analyse des HTML angepasst werden.
company_name = row.find_element(By.XPATH, ".//td[2]//a").text.strip() company_name = row.find_element(By.XPATH, ".//td[2]//a").text.strip()
website = row.find_element(By.XPATH, ".//td[3]//a").text.strip() website = row.find_element(By.XPATH, ".//td[3]//a").text.strip()
if company_name and website: if company_name and website:
results.append({'name': company_name, 'website': website}) results.append({'name': company_name, 'website': website})
except NoSuchElementException: except NoSuchElementException:
logger.warning("Konnte Name oder Webseite in einer Zeile nicht finden, überspringe Zeile.") logger.warning("Konnte Name oder Webseite in einer Zeile nicht finden, überspringe Zeile.")
continue continue
logger.info(f"{len(results)} Firmen erfolgreich extrahiert.") logger.info(f"{len(results)} Firmen erfolgreich extrahiert.")
return results return results
except Exception as e: except Exception as e: