diff --git a/scrape_fotograf.py b/scrape_fotograf.py index ee4ca71d..02f26f65 100644 --- a/scrape_fotograf.py +++ b/scrape_fotograf.py @@ -16,24 +16,30 @@ OUTPUT_DIR = 'output' OUTPUT_FILE = os.path.join(OUTPUT_DIR, 'nutzer_ohne_logins.csv') LOGIN_URL = 'https://app.fotograf.de/login/login' -# --- Selektoren --- -# Diese werden im nächsten Schritt korrigiert. Für diesen Test sind sie teilweise irrelevant. +# --- Selektoren (FINALE, KORREKTE VERSION) --- SELECTORS = { "cookie_accept_button": "#CybotCookiebotDialogBodyLevelButtonLevelOptinAllowAll", "login_user": "#login-email", "login_pass": "#login-password", "login_button": "#login-submit", "job_name": "h1", - # 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 + # Album-Übersicht (basierend auf Ihrem XPath, der enthielt) + "album_overview_rows": "//table/tbody/tr", + "album_overview_link": ".//td[2]//a", # Link ist in der 2. Spalte + "album_overview_logins": ".//td[7]", # Logins sind in der 7. Spalte + + # Einzelpersonen-Ansicht (innerhalb eines Albums) + "person_rows": "//section[.//h3[contains(., 'Einzelfotos')]]//table/tbody/tr", + "person_vorname": ".//td[4]", + "person_logins": ".//td[6]", + "person_buyer_link": ".//td[7]//a", + + # Käufer-Detailseite + "buyer_email": "//span[contains(., '@')]" } def take_error_screenshot(driver, error_name): + # ... (Funktion bleibt unverändert) ... os.makedirs(OUTPUT_DIR, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"error_{error_name}_{timestamp}.png" @@ -45,6 +51,7 @@ def take_error_screenshot(driver, error_name): print(f"!!! Konnte keinen Screenshot speichern: {e}") def setup_driver(): + # ... (Funktion bleibt unverändert) ... print("Initialisiere Chrome WebDriver...") options = Options() options.add_argument('--headless') @@ -52,7 +59,6 @@ def setup_driver(): options.add_argument('--disable-dev-shm-usage') options.add_argument('--window-size=1920,1200') options.binary_location = '/usr/bin/google-chrome' - try: driver = webdriver.Chrome(options=options) return driver @@ -60,14 +66,8 @@ def setup_driver(): print(f"Fehler bei der Initialisierung des WebDrivers: {e}") return None -def load_all_credentials(): - try: - with open(CREDENTIALS_FILE, 'r') as f: - return json.load(f) - except (FileNotFoundError, json.JSONDecodeError): - return None - def login(driver, username, password): + # ... (Funktion bleibt unverändert) ... print("Starte Login-Vorgang...") try: driver.get(LOGIN_URL) @@ -94,48 +94,102 @@ def login(driver, username, password): take_error_screenshot(driver, "login_error") return False -# 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 = [] +# Die finale, vollständige Logik +def process_full_job(driver, job_url): 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 + # 1. Job-Namen holen + print(f"\nVerarbeite Job-URL: {job_url}") + driver.get(job_url) + 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 [] + # 2. Alle Album-Links von der Übersichtsseite sammeln + job_id = job_url.split('/')[-1] + albums_overview_url = f"https://app.fotograf.de/config_jobs_photos/index/{job_id}" + print(f"Navigiere zur Alben-Übersicht: {albums_overview_url}") + driver.get(albums_overview_url) + + albums_to_visit = [] + try: + album_rows = wait.until(EC.presence_of_all_elements_located((By.XPATH, SELECTORS["album_overview_rows"]))) + print(f"{len(album_rows)} Alben in der Übersicht gefunden.") + for row in album_rows: + album_name = row.find_element(By.XPATH, SELECTORS["album_overview_link"]).text + album_link = row.find_element(By.XPATH, SELECTORS["album_overview_link"]).get_attribute('href') + albums_to_visit.append({"name": album_name, "url": album_link}) + print(f"Sammeln der Album-Links abgeschlossen.") + except TimeoutException: + print("Konnte die Album-Liste nicht finden.") + take_error_screenshot(driver, "album_overview_error") + return [] + + # 3. Jedes Album besuchen und die Personen mit 0 Logins finden + final_results = [] + for album in albums_to_visit: + print(f"\n--- Betrete Album: {album['name']} ---") + driver.get(album['url']) + try: + person_rows = wait.until(EC.presence_of_all_elements_located((By.XPATH, SELECTORS["person_rows"]))) + print(f"{len(person_rows)} Personen in diesem Album gefunden.") + + for person_row in person_rows: + try: + login_count = int(person_row.find_element(By.XPATH, SELECTORS["person_logins"]).text) + + if login_count == 0: + vorname = person_row.find_element(By.XPATH, SELECTORS["person_vorname"]).text + print(f" --> ERFOLG: '{vorname}' mit 0 Logins gefunden!") + + buyer_link_element = person_row.find_element(By.XPATH, SELECTORS["person_buyer_link"]) + buyer_page_url = buyer_link_element.get_attribute('href') + + # Temporär eine neue Seite öffnen, um die E-Mail zu holen + # (man könnte auch die buyer_page_url speichern und später abarbeiten) + current_window = driver.current_window_handle + driver.execute_script("window.open(arguments[0]);", buyer_page_url) + driver.switch_to.window(driver.window_handles[-1]) + + email = wait.until(EC.visibility_of_element_located((By.XPATH, SELECTORS["buyer_email"]))).text + print(f" E-Mail gefunden: {email}") + + final_results.append({ + "Auftragsname": job_name, + "Album": album['name'], + "Kind Vorname": vorname, + "Käufer E-Mail": email + }) + + # Tab schließen und zurückkehren + driver.close() + driver.switch_to.window(current_window) + + except (ValueError, NoSuchElementException): + # Ignoriere Zeilen, die nicht dem Format entsprechen + continue + except TimeoutException: + print(f" Keine Personen-Tabelle im Album '{album['name']}' gefunden. Überspringe.") + take_error_screenshot(driver, f"album_{album['name']}_error") + continue + + return final_results + +# ... (Rest des Skripts: save_results_to_csv, get_profile_choice, etc. bleiben gleich) ... def save_results_to_csv(results): if not results: print("\nKeine Daten zum Speichern vorhanden.") return os.makedirs(OUTPUT_DIR, exist_ok=True) + # Feldnamen an die neue Struktur anpassen + fieldnames = ["Auftragsname", "Album", "Kind Vorname", "Käufer E-Mail"] print(f"\nSpeichere {len(results)} Ergebnisse in '{OUTPUT_FILE}'...") with open(OUTPUT_FILE, 'w', newline='', encoding='utf-8') as f: - writer = csv.DictWriter(f, fieldnames=results[0].keys()) + writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(results) print("Speichern erfolgreich!") @@ -156,30 +210,21 @@ 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 (DIREKTER ALBUM-TEST) ---") + print("--- Fotograf.de Scraper für Nutzer ohne Logins (FINALE VERSION) ---") credentials = get_profile_choice() if not credentials: 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" + job_url = input("Bitte gib die URL des zu bearbeitenden Fotoauftrags ein (Einstellungs-Seite): ") + if "fotograf.de/config_jobs_settings/index/" not in job_url: + print("Dies scheint keine gültige URL für die Auftragseinstellungen zu sein.") + return driver = setup_driver() if not driver: return try: if login(driver, credentials['username'], credentials['password']): - # 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) + all_results = process_full_job(driver, job_url) save_results_to_csv(all_results) else: print("Skript wird beendet, da der Login fehlgeschlagen ist.")