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-11 07:42:19 +00:00
parent 5f90704308
commit 13e3efa95d
1 changed files with 33 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
dealfront_enrichment.py
View File

@@ -21,7 +21,7 @@ class Config:
# --- 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) # Reduziert Selenium-Spam
logging.getLogger("selenium.webdriver.remote").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
os.makedirs(Config.OUTPUT_DIR, exist_ok=True)
@@ -38,7 +38,7 @@ class DealfrontScraper:
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,1080")
chrome_options.add_argument("--window-size=1920,1200")
try:
self.driver = webdriver.Chrome(options=chrome_options)
@@ -48,11 +48,8 @@ class DealfrontScraper:
self.wait = WebDriverWait(self.driver, 30)
self.username, self.password = self._load_credentials()
# FAIL-FAST: Sofortiger Abbruch, wenn Credentials fehlen
if not self.username or not self.password:
raise ValueError("Benutzername oder Passwort konnten nicht aus der Credentials-Datei geladen werden. Breche ab.")
raise ValueError("Credentials konnten nicht geladen werden. Breche ab.")
logger.info("WebDriver erfolgreich initialisiert.")
def _load_credentials(self):
@@ -60,73 +57,65 @@ class DealfrontScraper:
with open(Config.CREDENTIALS_FILE, 'r', encoding='utf-8') as f:
creds = json.load(f)
return creds.get("username"), creds.get("password")
except Exception as e:
logger.error(f"Credentials-Datei {Config.CREDENTIALS_FILE} konnte nicht geladen werden: {e}")
except Exception:
logger.error(f"Credentials-Datei {Config.CREDENTIALS_FILE} nicht gefunden.")
return None, None
def _save_debug_artifacts(self):
def _save_debug_artifacts(self, suffix=""):
try:
timestamp = time.strftime("%Y%m%d-%H%M%S")
screenshot_path = os.path.join(Config.OUTPUT_DIR, f"error_{timestamp}.png")
html_path = os.path.join(Config.OUTPUT_DIR, f"error_{timestamp}.html")
self.driver.save_screenshot(screenshot_path)
logger.error(f"Debug-Screenshot gespeichert: {screenshot_path}")
with open(html_path, "w", encoding="utf-8") as f:
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-HTML-Quellcode gespeichert: {html_path}")
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 zur Login-Seite: {Config.LOGIN_URL}")
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. Kurze Pause für die Weiterleitung.")
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 die Suche: '{Config.SEARCH_NAME}'")
logger.info(f"Navigiere direkt zur Target-Seite und lade Suche: '{Config.SEARCH_NAME}'")
self.driver.get(Config.TARGET_URL)
self.wait.until(EC.url_contains("/t/prospector/"))
self.wait.until(EC.element_to_be_clickable((By.XPATH, f"//*[normalize-space()='{Config.SEARCH_NAME}']"))).click()
search_item_selector = (By.XPATH, f"//div[contains(@class, 'truncate') and normalize-space()='{Config.SEARCH_NAME}']")
self.wait.until(EC.element_to_be_clickable(search_item_selector)).click()
# 3. ERGEBNISSE EXTRAHIEREN
logger.info("Suche geladen. Extrahiere Ergebnisse der ersten Seite.")
results_table_selector = (By.CSS_SELECTOR, "table#t-result-table tbody tr[id]")
self.wait.until(EC.presence_of_element_located(results_table_selector))
data_rows = self.driver.find_elements(By.XPATH, "//tr[.//a[contains(@class, 't-highlight-text')]]")
logger.info(f"{len(data_rows)} gültige Datenzeilen gefunden.")
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
companies = []
for row in data_rows:
try:
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()
companies.append({'name': name, 'website': website})
except NoSuchElementException:
continue
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")
return companies
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 close(self):
if self.driver:
self.driver.quit()
logger.info("WebDriver geschlossen.")
if self.driver: self.driver.quit()
if __name__ == "__main__":
scraper = None
try:
scraper = DealfrontScraper()
company_list = scraper.run()
companies = scraper.run()
if company_list:
df = pd.DataFrame(company_list)
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', 120)
@@ -137,10 +126,10 @@ if __name__ == "__main__":
print("="*80)
print(df.to_string(index=False))
print("="*80 + "\n")
logger.info("Workflow erfolgreich abgeschlossen.")
else:
logger.warning("Keine Firmen konnten extrahiert werden.")
logger.warning("Keine Firmen konnten extrahiert werden. Bitte Debug-Artefakte prüfen.")
logger.info("Workflow erfolgreich abgeschlossen.")
except Exception as e:
logger.critical(f"Ein kritischer Fehler ist im Hauptprozess aufgetreten: {e}", exc_info=False)
finally: