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

scrape_fotograf.py aktualisiert

Browse Source
This commit is contained in:
Floke
2025-07-16 15:36:07 +00:00
parent 471be85c97
commit 8a42c27942
1 changed files with 23 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
scrape_fotograf.py
View File

@@ -8,7 +8,7 @@ from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By from selenium.webdriver.common.by import By
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, StaleElementReferenceException
# --- Konfiguration & Konstanten --- # --- Konfiguration & Konstanten ---
CREDENTIALS_FILE = 'fotograf_credentials.json' CREDENTIALS_FILE = 'fotograf_credentials.json'
@@ -16,30 +16,24 @@ OUTPUT_DIR = 'output'
OUTPUT_FILE = os.path.join(OUTPUT_DIR, 'nutzer_ohne_logins.csv') OUTPUT_FILE = os.path.join(OUTPUT_DIR, 'nutzer_ohne_logins.csv')
LOGIN_URL = 'https://app.fotograf.de/login/login' LOGIN_URL = 'https://app.fotograf.de/login/login'
# --- Selektoren (FINALE VERSION 3.0) --- # --- Selektoren (unverändert) ---
SELECTORS = { SELECTORS = {
"cookie_accept_button": "#CybotCookiebotDialogBodyLevelButtonLevelOptinAllowAll", "cookie_accept_button": "#CybotCookiebotDialogBodyLevelButtonLevelOptinAllowAll",
"login_user": "#login-email", "login_user": "#login-email",
"login_pass": "#login-password", "login_pass": "#login-password",
"login_button": "#login-submit", "login_button": "#login-submit",
"job_name": "h1", "job_name": "h1",
# Album-Übersicht
"album_overview_rows": "//table/tbody/tr", "album_overview_rows": "//table/tbody/tr",
"album_overview_link": ".//td[2]//a", "album_overview_link": ".//td[2]//a",
# Einzelpersonen-Ansicht (innerhalb eines Albums)
"person_rows": "//div[contains(@class, 'border-legacy-silver-550') and .//span[text()='Logins']]", "person_rows": "//div[contains(@class, 'border-legacy-silver-550') and .//span[text()='Logins']]",
"person_vorname": ".//span[text()='Vorname']/following-sibling::strong", "person_vorname": ".//span[text()='Vorname']/following-sibling::strong",
"person_logins": ".//span[text()='Logins']/following-sibling::strong", "person_logins": ".//span[text()='Logins']/following-sibling::strong",
"person_access_code_link": ".//a[contains(@data-qa-id, 'guest-access-banner-access-code')]", "person_access_code_link": ".//a[contains(@data-qa-id, 'guest-access-banner-access-code')]",
# "Kommunikation"-Seite (nach Klick auf Zugangscode)
"potential_buyer_link": "//a[contains(@href, '/config_customers/view_customer')]", "potential_buyer_link": "//a[contains(@href, '/config_customers/view_customer')]",
# Käufer-Detailseite (finale Seite)
"buyer_email": "//span[contains(., '@')]" "buyer_email": "//span[contains(., '@')]"
} }
# ... (Funktionen take_error_screenshot, setup_driver, load_all_credentials, login bleiben unverändert) ...
def take_error_screenshot(driver, error_name): def take_error_screenshot(driver, error_name):
os.makedirs(OUTPUT_DIR, exist_ok=True) os.makedirs(OUTPUT_DIR, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -99,6 +93,7 @@ def login(driver, username, password):
take_error_screenshot(driver, "login_error") take_error_screenshot(driver, "login_error")
return False return False
def process_full_job(driver, job_url): def process_full_job(driver, job_url):
wait = WebDriverWait(driver, 15) wait = WebDriverWait(driver, 15)
@@ -148,18 +143,28 @@ def process_full_job(driver, job_url):
vorname = person_row.find_element(By.XPATH, SELECTORS["person_vorname"]).text vorname = person_row.find_element(By.XPATH, SELECTORS["person_vorname"]).text
print(f" --> ERFOLG: '{vorname}' mit 0 Logins gefunden!") print(f" --> ERFOLG: '{vorname}' mit 0 Logins gefunden!")
# 1. Gehe zur "Kommunikation"-Seite
access_code_page_url = person_row.find_element(By.XPATH, SELECTORS["person_access_code_link"]).get_attribute('href') access_code_page_url = person_row.find_element(By.XPATH, SELECTORS["person_access_code_link"]).get_attribute('href')
driver.get(access_code_page_url) driver.get(access_code_page_url)
print(f" Navigiere zur Kommunikations-Seite für '{vorname}'...") print(f" Navigiere zur Kommunikations-Seite für '{vorname}'...")
# 2. Klicke auf den Link zum potenziellen Käufer, um zur finalen Seite zu gelangen # ----- NEUER, ROBUSTER RETRY-BLOCK -----
wait.until(EC.element_to_be_clickable((By.XPATH, SELECTORS["potential_buyer_link"]))).click() # Dieser Block versucht den Klick bis zu 3 Mal, falls ein StaleElement-Fehler auftritt.
print(f" Navigiere zur Käufer-Detailseite...") for attempt in range(3):
try:
wait.until(EC.element_to_be_clickable((By.XPATH, SELECTORS["potential_buyer_link"]))).click()
print(f" Navigiere zur Käufer-Detailseite...")
break # Erfolgreich, also Schleife verlassen
except StaleElementReferenceException:
print(f" Timing-Fehler (StaleElement), Versuch {attempt + 1}/3...")
time.sleep(0.5) # Kurze Pause vor dem nächsten Versuch
else:
# Wird ausgeführt, wenn die Schleife ohne 'break' endet
print(" Konnte Käufer-Link auch nach mehreren Versuchen nicht klicken.")
raise # Den letzten Fehler erneut auslösen, um das Problem zu signalisieren
# ----- ENDE DES RETRY-BLOCKS -----
# 3. Extrahiere die E-Mail
email = wait.until(EC.visibility_of_element_located((By.XPATH, SELECTORS["buyer_email"]))).text email = wait.until(EC.visibility_of_element_located((By.XPATH, SELECTORS["buyer_email"]))).text
print(f" E-Mail gefunden: {email}") print(f" FINALE ERFOLG: E-Mail gefunden: {email}")
final_results.append({ final_results.append({
"Auftragsname": job_name, "Auftragsname": job_name,
@@ -168,10 +173,8 @@ def process_full_job(driver, job_url):
"Käufer E-Mail": email "Käufer E-Mail": email
}) })
# Gehe zurück zur Album-Detailseite, um die nächste Person zu prüfen driver.get(album['url']) # Zurück zur Album-Detailseite
driver.get(album['url']) except (ValueError, NoSuchElementException, TimeoutException):
except (ValueError, NoSuchElementException, TimeoutException) as e:
#print(f" Fehler beim Verarbeiten einer Person: {e}")
continue continue
except TimeoutException: except TimeoutException:
print(f" Keine Personen-Daten im Album '{album['name']}' gefunden. Überspringe.") print(f" Keine Personen-Daten im Album '{album['name']}' gefunden. Überspringe.")
@@ -180,6 +183,7 @@ def process_full_job(driver, job_url):
return final_results return final_results
# ... (Funktionen save_results_to_csv, get_profile_choice, main bleiben unverändert) ...
def save_results_to_csv(results): def save_results_to_csv(results):
if not results: if not results:
print("\nKeine Daten zum Speichern vorhanden.") print("\nKeine Daten zum Speichern vorhanden.")
@@ -194,7 +198,6 @@ def save_results_to_csv(results):
print("Speichern erfolgreich!") print("Speichern erfolgreich!")
def get_profile_choice(): def get_profile_choice():
# ... unverändert ...
all_credentials = load_all_credentials() all_credentials = load_all_credentials()
if not all_credentials: return None if not all_credentials: return None
profiles = list(all_credentials.keys()) profiles = list(all_credentials.keys())
@@ -211,7 +214,6 @@ def get_profile_choice():
except ValueError: print("Ungültige Eingabe.") except ValueError: print("Ungültige Eingabe.")
def main(): def main():
# ... unverändert ...
print("--- Fotograf.de Scraper für Nutzer ohne Logins (FINALE VERSION) ---") print("--- Fotograf.de Scraper für Nutzer ohne Logins (FINALE VERSION) ---")
credentials = get_profile_choice() credentials = get_profile_choice()
if not credentials: return if not credentials: return