From 7d8398280a20426bce8e5f2454df08254b39736b Mon Sep 17 00:00:00 2001 From: Floke Date: Wed, 16 Jul 2025 15:04:44 +0000 Subject: [PATCH] scrape_fotograf.py aktualisiert --- scrape_fotograf.py | 147 ++++++++++++++++++--------------------------- 1 file changed, 58 insertions(+), 89 deletions(-) diff --git a/scrape_fotograf.py b/scrape_fotograf.py index 17a41fcb..ee4ca71d 100644 --- a/scrape_fotograf.py +++ b/scrape_fotograf.py @@ -17,19 +17,20 @@ OUTPUT_FILE = os.path.join(OUTPUT_DIR, 'nutzer_ohne_logins.csv') LOGIN_URL = 'https://app.fotograf.de/login/login' # --- Selektoren --- -# FINALE VERSION mit XPath für Textsuche +# Diese werden im nächsten Schritt korrigiert. Für diesen Test sind sie teilweise irrelevant. SELECTORS = { "cookie_accept_button": "#CybotCookiebotDialogBodyLevelButtonLevelOptinAllowAll", "login_user": "#login-email", "login_pass": "#login-password", "login_button": "#login-submit", "job_name": "h1", - # NEU: XPath, um die Zeilen nach der Kopfzeile zu finden. - "album_rows": "//div[div[contains(., 'Fotos insgesamt')]]/following-sibling::div", - "album_link": ".//a[contains(@href, '/config_jobs_photos/gallery/')]", # .// sucht nur innerhalb des Kontexts - "login_count": ".//div[7]", - "buyer_link": "a.block:has(span:contains('Käufer'))", - "buyer_email": "div.flex:nth-of-type(4) span" + # Platzhalter, wird im nächsten Schritt korrigiert + "album_rows": "//table/tbody/tr", + "album_link": ".//td[2]//a", + "login_count": ".//td[7]", + # Diese sind für den Test relevant + "buyer_link": "//a[contains(., 'Käufer')]", # Robusterer XPath + "buyer_email": "//span[contains(., '@')]" # Robusterer XPath } def take_error_screenshot(driver, error_name): @@ -88,86 +89,43 @@ def login(driver, username, password): wait.until(EC.url_contains('/config_dashboard/index')) print("Login erfolgreich!") return True - except TimeoutException: - print("Login fehlgeschlagen.") - take_error_screenshot(driver, "login_timeout") - return False except Exception as e: - print(f"Ein unerwarteter Fehler beim Login: {e}") - take_error_screenshot(driver, "login_unexpected") + print(f"Login fehlgeschlagen. Grund: {e}") + take_error_screenshot(driver, "login_error") return False -def process_job(driver, job_url): - print(f"\nVerarbeite Job-URL: {job_url}") - job_id = job_url.split('/')[-1] - albums_url = f"https://app.fotograf.de/config_jobs_photos/index/{job_id}" - settings_url = f"https://app.fotograf.de/config_jobs_settings/index/{job_id}" - - driver.get(settings_url) - wait = WebDriverWait(driver, 15) - - try: - job_name = wait.until(EC.visibility_of_element_located((By.CSS_SELECTOR, SELECTORS["job_name"]))).text - print(f"Auftragsname: '{job_name}'") - except TimeoutException: - print("Konnte den Auftragsnamen nicht finden.") - take_error_screenshot(driver, "job_name_not_found") - return [] - - print(f"Navigiere zur Alben-Übersicht: {albums_url}") - driver.get(albums_url) - - albums_to_process = [] - try: - # GEÄNDERT: Warten auf die Kopfzeile mit einem GÜLTIGEN XPath Selector - header_xpath = "//div[contains(., 'Fotos insgesamt')]" - print(f"Warte auf Tabellen-Kopfzeile mit XPath: {header_xpath}") - wait.until(EC.presence_of_element_located((By.XPATH, header_xpath))) - time.sleep(1) - - # GEÄNDERT: Finde die Album-Zeilen mit dem GÜLTIGEN XPath Selector - album_rows = driver.find_elements(By.XPATH, SELECTORS["album_rows"]) - print(f"{len(album_rows)} Album-Zeilen gefunden. Prüfe auf Logins...") - - for i, row in enumerate(album_rows): - print(f"\n--- Analysiere Zeile {i+1} ---") - try: - row_html = row.get_attribute('outerHTML') - print(f"DEBUG (HTML-Ausschnitt): {row_html[:400]}...") - - # GEÄNDERT: XPath-Selektoren im Kontext der Zeile verwenden - login_count_element = row.find_element(By.XPATH, SELECTORS["login_count"]) - login_count_text = login_count_element.text.strip() - - print(f"DEBUG (Gefundener Login-Text): '{login_count_text}'") - - if int(login_count_text) == 0: - album_link_element = row.find_element(By.XPATH, SELECTORS["album_link"]) - child_name = album_link_element.text - album_link = album_link_element.get_attribute('href') - - albums_to_process.append({ - "child_name": child_name, - "album_detail_url": album_link - }) - print(f" --> ERFOLG: Album '{child_name}' mit 0 Logins zur Verarbeitung hinzugefügt.") - else: - print(f" --> INFO: Album wird übersprungen (Logins > 0).") - - except (NoSuchElementException, ValueError) as e: - print(f" --> FEHLER: Konnte Zeile nicht verarbeiten. Grund: {e}") - - except (TimeoutException, InvalidSelectorException) as e: - print(f"Die Album-Tabelle wurde nicht gefunden oder der Selector war ungültig. Fehler: {e}") - take_error_screenshot(driver, "album_list_timeout") - return [] - +# NEU: Eine dedizierte Funktion, um nur die Details einer Album-URL zu verarbeiten +def process_single_album_details(driver, album_url, job_name): + """Ruft eine Album-Detailseite auf und extrahiert die Käufer-Infos.""" results = [] - print(f"\nVerarbeite {len(albums_to_process)} Alben mit 0 Logins im Detail...") - for album in albums_to_process: - # ... Rest der Funktion bleibt gleich ... - pass - + wait = WebDriverWait(driver, 15) + try: + print(f"\n--> TEST: Rufe Album-Detailseite direkt auf: {album_url}") + driver.get(album_url) + + print("Suche nach Käufer-Link...") + buyer_link_element = wait.until(EC.visibility_of_element_located((By.XPATH, SELECTORS["buyer_link"]))) + buyer_name = buyer_link_element.text.replace('Käufer ', '').strip() + buyer_page_url = buyer_link_element.get_attribute('href') + + print(f" Käufer gefunden: '{buyer_name}'. Rufe Käuferseite auf...") + driver.get(buyer_page_url) + + print("Suche nach E-Mail-Adresse...") + buyer_email = wait.until(EC.visibility_of_element_located((By.XPATH, SELECTORS["buyer_email"]))).text + print(f" E-Mail gefunden: {buyer_email}") + + results.append({ + "Auftragsname": job_name, + "Kind Vorname": "Kobolde (Test)", # Platzhalter, da wir die Liste nicht scannen + "Käufer Name": buyer_name, + "Käufer E-Mail": buyer_email, + }) + print("\n--> ERFOLG: Käufer-Daten erfolgreich extrahiert!") + except Exception as e: + print(f"--> FEHLER: Konnte Käufer-Details nicht extrahieren. Grund: {e}") + take_error_screenshot(driver, "detail_page_error") + return results def save_results_to_csv(results): @@ -198,19 +156,30 @@ def get_profile_choice(): else: print("Ungültige Auswahl.") except ValueError: print("Ungültige Eingabe.") +# GEÄNDERT: Die main-Funktion führt jetzt den direkten Test aus def main(): - print("--- Fotograf.de Scraper für Nutzer ohne Logins ---") + print("--- Fotograf.de Scraper für Nutzer ohne Logins (DIREKTER ALBUM-TEST) ---") credentials = get_profile_choice() if not credentials: return - job_url = input("Bitte gib die URL des zu bearbeitenden Fotoauftrags ein: ") - if "fotograf.de/config_jobs_settings/index/" not in job_url: - print("Dies scheint keine gültige URL zu sein.") - return + + # Die Job-URL wird nur noch für den Namen des Auftrags benötigt + job_settings_url = "https://app.fotograf.de/config_jobs_settings/index/216142382" + # Die von Ihnen bereitgestellte Test-URL + test_album_url = "https://app.fotograf.de/config_jobs_photos/gallery/216142382/6137045" + driver = setup_driver() if not driver: return + try: if login(driver, credentials['username'], credentials['password']): - all_results = process_job(driver, job_url) + # Job-Namen holen + print(f"Rufe Job-Seite für Auftragsnamen auf: {job_settings_url}") + driver.get(job_settings_url) + job_name = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.CSS_SELECTOR, SELECTORS["job_name"]))).text + print(f"Auftragsname gefunden: '{job_name}'") + + # Führe den Test für das einzelne Album aus + all_results = process_single_album_details(driver, test_album_url, job_name) save_results_to_csv(all_results) else: print("Skript wird beendet, da der Login fehlgeschlagen ist.")