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:04:44 +00:00
parent b90c6d6c33
commit eb9b47ffe0
1 changed files with 58 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

147
scrape_fotograf.py
View File

@@ -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.")