#!/usr/bin/env python3 """ Eurojackpot Historical Data Updater Lädt automatisch die neuesten Eurojackpot-Ziehungen von der offiziellen Website und aktualisiert die lokale CSV-Datei. Quellen: - https://www.eurojackpot.de/de/eurojackpot/gewinnzahlen.html - Oder alternative APIs/Websites Features: - Automatisches Scraping der neuesten Ziehungen - Duplikate-Vermeidung - Backup vor Update - Validierung der neuen Daten """ import pandas as pd import requests from bs4 import BeautifulSoup from datetime import datetime import shutil import os import re from typing import List, Dict, Optional import time class EurojackpotDataUpdater: """Aktualisiert historische Eurojackpot-Daten automatisch.""" def __init__(self, data_file: str): """ Initialisiert den Updater. Args: data_file: Pfad zur lokalen CSV-Datei """ self.data_file = data_file self.backup_file = None self.df_existing = None # API/Scraping URLs self.sources = { 'eurojackpot_de': 'https://www.eurojackpot.de/de/eurojackpot/gewinnzahlen.html', 'euro-jackpot_net': 'https://www.euro-jackpot.net/de/gewinnzahlen', 'lottode': 'https://www.lotto.de/eurojackpot/gewinnzahlen' } print("🔄 EUROJACKPOT DATA UPDATER") print("=" * 60) def load_existing_data(self) -> bool: """Lädt existierende Daten.""" try: if not os.path.exists(self.data_file): print(f"⚠️ Datei nicht gefunden: {self.data_file}") print(" Erstelle neue Datei...") self.df_existing = pd.DataFrame(columns=['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']) return True self.df_existing = pd.read_csv(self.data_file, sep=';') # Datum als datetime if 'datum' in self.df_existing.columns: self.df_existing['datum'] = pd.to_datetime( self.df_existing['datum'], format='%Y-%m-%d', errors='coerce' ) print(f"✅ Existierende Daten geladen: {len(self.df_existing)} Ziehungen") if len(self.df_existing) > 0: latest = self.df_existing['datum'].max() print(f" Neueste Ziehung: {latest.strftime('%Y-%m-%d') if pd.notna(latest) else 'Unbekannt'}") return True except Exception as e: print(f"❌ Fehler beim Laden: {e}") return False def create_backup(self) -> bool: """Erstellt Backup der existierenden Datei.""" if not os.path.exists(self.data_file): return True try: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_dir = os.path.join(os.path.dirname(self.data_file), "backups") os.makedirs(backup_dir, exist_ok=True) filename = os.path.basename(self.data_file) self.backup_file = os.path.join(backup_dir, f"{filename}.backup_{timestamp}") shutil.copy2(self.data_file, self.backup_file) print(f"✅ Backup erstellt: {os.path.basename(self.backup_file)}") return True except Exception as e: print(f"⚠️ Backup-Fehler: {e}") return False def fetch_from_eurojackpot_de(self) -> List[Dict]: """ Scrapt Daten von eurojackpot.de Returns: Liste von Ziehungen als Dictionaries """ print("\n🌐 Versuche eurojackpot.de...") try: headers = { 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36' } response = requests.get(self.sources['eurojackpot_de'], headers=headers, timeout=10) response.raise_for_status() soup = BeautifulSoup(response.content, 'html.parser') # Suche nach Ziehungsergebnissen # Dies ist ein generisches Beispiel - muss an tatsächliche Website-Struktur angepasst werden draws = [] # Beispiel-Parsing (muss angepasst werden!) result_blocks = soup.find_all('div', class_='drawing-result') for block in result_blocks[:10]: # Letzte 10 Ziehungen try: # Datum extrahieren date_elem = block.find('span', class_='date') if not date_elem: continue date_str = date_elem.text.strip() date_obj = self._parse_german_date(date_str) # Zahlen extrahieren numbers = [] number_elems = block.find_all('span', class_='ball') for num_elem in number_elems[:5]: num = int(num_elem.text.strip()) numbers.append(num) # Eurozahlen extrahieren euro_numbers = [] euro_elems = block.find_all('span', class_='euro-ball') for euro_elem in euro_elems[:2]: euro_num = int(euro_elem.text.strip()) euro_numbers.append(euro_num) if len(numbers) == 5 and len(euro_numbers) == 2: draws.append({ 'datum': date_obj, 'Z1': numbers[0], 'Z2': numbers[1], 'Z3': numbers[2], 'Z4': numbers[3], 'Z5': numbers[4], 'SZ1': euro_numbers[0], 'SZ2': euro_numbers[1] }) except Exception as e: continue if draws: print(f" ✅ {len(draws)} Ziehungen gefunden") return draws else: print(" ⚠️ Keine Ziehungen gefunden (HTML-Struktur möglicherweise geändert)") return [] except Exception as e: print(f" ❌ Fehler: {e}") return [] def fetch_from_euro_jackpot_net(self) -> List[Dict]: """ Scrapt Daten von euro-jackpot.net Returns: Liste von Ziehungen """ print("\n🌐 Versuche euro-jackpot.net...") try: headers = { 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36' } response = requests.get(self.sources['euro-jackpot_net'], headers=headers, timeout=10) response.raise_for_status() soup = BeautifulSoup(response.content, 'html.parser') draws = [] # Parsing-Logik für diese Website # (Platzhalter - muss an tatsächliche Struktur angepasst werden) result_rows = soup.find_all('tr', class_='result-row') for row in result_rows[:10]: try: cells = row.find_all('td') if len(cells) < 8: continue # Datum date_str = cells[0].text.strip() date_obj = self._parse_german_date(date_str) # Hauptzahlen numbers = [int(cells[i].text.strip()) for i in range(1, 6)] # Eurozahlen euro_numbers = [int(cells[i].text.strip()) for i in range(6, 8)] draws.append({ 'datum': date_obj, 'Z1': numbers[0], 'Z2': numbers[1], 'Z3': numbers[2], 'Z4': numbers[3], 'Z5': numbers[4], 'SZ1': euro_numbers[0], 'SZ2': euro_numbers[1] }) except Exception as e: continue if draws: print(f" ✅ {len(draws)} Ziehungen gefunden") else: print(" ⚠️ Keine Ziehungen gefunden") return draws except Exception as e: print(f" ❌ Fehler: {e}") return [] def fetch_from_manual_input(self) -> List[Dict]: """ Manuelle Eingabe von neuen Ziehungen. Returns: Liste von Ziehungen """ print("\n⌨️ MANUELLE EINGABE") print("=" * 60) print("Gib die neuesten Ziehungen manuell ein.") print("Format: YYYY-MM-DD Z1 Z2 Z3 Z4 Z5 SZ1 SZ2") print("Beispiel: 2025-01-24 7 18 26 37 46 3 11") print("Leer lassen zum Beenden.\n") draws = [] while True: user_input = input(f"Ziehung {len(draws) + 1}: ").strip() if not user_input: break try: parts = user_input.split() if len(parts) != 8: print(" ❌ Ungültiges Format. Bitte 8 Werte eingeben.") continue date_obj = datetime.strptime(parts[0], '%Y-%m-%d') numbers = [int(parts[i]) for i in range(1, 6)] euro_numbers = [int(parts[i]) for i in range(6, 8)] # Validierung if not all(1 <= n <= 50 for n in numbers): print(" ❌ Hauptzahlen müssen zwischen 1 und 50 liegen.") continue if not all(1 <= n <= 12 for n in euro_numbers): print(" ❌ Eurozahlen müssen zwischen 1 und 12 liegen.") continue if len(set(numbers)) != 5: print(" ❌ Hauptzahlen müssen eindeutig sein.") continue if len(set(euro_numbers)) != 2: print(" ❌ Eurozahlen müssen eindeutig sein.") continue draws.append({ 'datum': date_obj, 'Z1': numbers[0], 'Z2': numbers[1], 'Z3': numbers[2], 'Z4': numbers[3], 'Z5': numbers[4], 'SZ1': euro_numbers[0], 'SZ2': euro_numbers[1] }) print(f" ✅ Ziehung hinzugefügt: {date_obj.strftime('%Y-%m-%d')}") except ValueError as e: print(f" ❌ Fehler: {e}") continue if draws: print(f"\n✅ {len(draws)} Ziehungen manuell eingegeben") return draws def _parse_german_date(self, date_str: str) -> datetime: """ Parst deutsches Datumsformat. Args: date_str: Datum als String (z.B. "24.01.2025" oder "24. Januar 2025") Returns: datetime Objekt """ # Entferne zusätzliche Leerzeichen date_str = re.sub(r'\s+', ' ', date_str.strip()) # Monatsnamen-Mapping months_de = { 'januar': 1, 'februar': 2, 'märz': 3, 'april': 4, 'mai': 5, 'juni': 6, 'juli': 7, 'august': 8, 'september': 9, 'oktober': 10, 'november': 11, 'dezember': 12 } # Versuche verschiedene Formate formats = [ '%d.%m.%Y', '%d.%m.%y', '%Y-%m-%d', '%d/%m/%Y' ] for fmt in formats: try: return datetime.strptime(date_str, fmt) except ValueError: continue # Versuche mit Monatsnamen for month_name, month_num in months_de.items(): if month_name.lower() in date_str.lower(): # Extrahiere Tag und Jahr match = re.search(r'(\d{1,2})\.?\s+' + month_name + r'\s+(\d{4})', date_str, re.IGNORECASE) if match: day = int(match.group(1)) year = int(match.group(2)) return datetime(year, month_num, day) # Fallback: aktuelles Datum print(f" ⚠️ Konnte Datum nicht parsen: {date_str}, nutze aktuelles Datum") return datetime.now() def fetch_new_draws(self, source: str = 'auto') -> List[Dict]: """ Holt neue Ziehungen von der gewählten Quelle. Args: source: 'auto', 'eurojackpot_de', 'euro_jackpot_net', 'manual' Returns: Liste von neuen Ziehungen """ print(f"\n🔍 SUCHE NACH NEUEN ZIEHUNGEN (Quelle: {source})") print("=" * 60) all_draws = [] if source == 'auto': # Versuche alle Quellen sources_to_try = [ ('eurojackpot_de', self.fetch_from_eurojackpot_de), ('euro_jackpot_net', self.fetch_from_euro_jackpot_net) ] for source_name, fetch_func in sources_to_try: draws = fetch_func() if draws: all_draws.extend(draws) break # Erste erfolgreiche Quelle nutzen time.sleep(1) # Pause zwischen Requests if not all_draws: print("\n⚠️ Automatisches Scraping fehlgeschlagen.") print(" Möchtest du Daten manuell eingeben? (j/n): ", end='') if input().lower() in ['j', 'ja', 'y', 'yes']: all_draws = self.fetch_from_manual_input() elif source == 'eurojackpot_de': all_draws = self.fetch_from_eurojackpot_de() elif source == 'euro_jackpot_net': all_draws = self.fetch_from_euro_jackpot_net() elif source == 'manual': all_draws = self.fetch_from_manual_input() else: print(f"❌ Unbekannte Quelle: {source}") return all_draws def validate_draw(self, draw: Dict) -> bool: """ Validiert eine Ziehung. Args: draw: Ziehungs-Dictionary Returns: True wenn valide """ try: # Prüfe Pflichtfelder required_fields = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2'] if not all(field in draw for field in required_fields): return False # Prüfe Hauptzahlen (1-50) main_numbers = [draw[f'Z{i}'] for i in range(1, 6)] if not all(1 <= n <= 50 for n in main_numbers): return False if len(set(main_numbers)) != 5: # Eindeutig return False # Prüfe Eurozahlen (1-12) euro_numbers = [draw['SZ1'], draw['SZ2']] if not all(1 <= n <= 12 for n in euro_numbers): return False if len(set(euro_numbers)) != 2: # Eindeutig return False # Prüfe Datum if not isinstance(draw['datum'], datetime): return False return True except Exception: return False def merge_with_existing(self, new_draws: List[Dict]) -> pd.DataFrame: """ Merged neue Ziehungen mit existierenden Daten. Args: new_draws: Liste von neuen Ziehungen Returns: Zusammengeführter DataFrame """ print(f"\n🔀 MERGE MIT EXISTIERENDEN DATEN") print("=" * 60) # Validiere neue Ziehungen valid_draws = [draw for draw in new_draws if self.validate_draw(draw)] if len(valid_draws) < len(new_draws): invalid_count = len(new_draws) - len(valid_draws) print(f"⚠️ {invalid_count} ungültige Ziehung(en) übersprungen") if not valid_draws: print("❌ Keine gültigen neuen Ziehungen zum Hinzufügen") return self.df_existing # Erstelle DataFrame aus neuen Ziehungen df_new = pd.DataFrame(valid_draws) # Kombiniere if self.df_existing is None or len(self.df_existing) == 0: df_combined = df_new else: df_combined = pd.concat([self.df_existing, df_new], ignore_index=True) # Entferne Duplikate (basierend auf Datum + Zahlen) before_dedup = len(df_combined) df_combined = df_combined.drop_duplicates( subset=['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5'], keep='first' ) after_dedup = len(df_combined) duplicates_removed = before_dedup - after_dedup if duplicates_removed > 0: print(f"🗑️ {duplicates_removed} Duplikat(e) entfernt") # Sortiere nach Datum df_combined = df_combined.sort_values('datum', ascending=True) df_combined = df_combined.reset_index(drop=True) # Berechne neue Einträge new_entries = len(df_combined) - len(self.df_existing) if self.df_existing is not None else len(df_combined) print(f"✅ Merge abgeschlossen:") print(f" Vorher: {len(self.df_existing) if self.df_existing is not None else 0} Ziehungen") print(f" Neu hinzugefügt: {new_entries} Ziehungen") print(f" Nachher: {len(df_combined)} Ziehungen") return df_combined def save_updated_data(self, df: pd.DataFrame) -> bool: """ Speichert aktualisierte Daten. Args: df: DataFrame zum Speichern Returns: True bei Erfolg """ try: # Format Datum als String df_to_save = df.copy() df_to_save['datum'] = df_to_save['datum'].dt.strftime('%Y-%m-%d') # Speichern df_to_save.to_csv(self.data_file, sep=';', index=False) print(f"\n💾 Daten gespeichert: {self.data_file}") print(f" {len(df)} Ziehungen total") if len(df) > 0: latest = df['datum'].max() print(f" Neueste Ziehung: {latest.strftime('%Y-%m-%d')}") return True except Exception as e: print(f"❌ Fehler beim Speichern: {e}") # Restore backup if self.backup_file and os.path.exists(self.backup_file): print("🔄 Stelle Backup wieder her...") shutil.copy2(self.backup_file, self.data_file) print("✅ Backup wiederhergestellt") return False def update(self, source: str = 'auto', dry_run: bool = False) -> bool: """ Führt komplettes Update durch. Args: source: Datenquelle ('auto', 'eurojackpot_de', 'euro_jackpot_net', 'manual') dry_run: Wenn True, keine Änderungen speichern Returns: True bei Erfolg """ print(f"\n{'🧪 DRY RUN MODE' if dry_run else '🚀 UPDATE STARTEN'}") print("=" * 60) # 1. Lade existierende Daten if not self.load_existing_data(): return False # 2. Backup erstellen if not dry_run: self.create_backup() # 3. Hole neue Ziehungen new_draws = self.fetch_new_draws(source=source) if not new_draws: print("\n✅ Keine neuen Ziehungen gefunden - Daten sind aktuell") return True # 4. Merge mit existierenden Daten df_updated = self.merge_with_existing(new_draws) # 5. Speichern if not dry_run: return self.save_updated_data(df_updated) else: print("\n🧪 DRY RUN - Keine Änderungen gespeichert") print(f" Würde {len(df_updated)} Ziehungen speichern") return True def main(): """Hauptfunktion für interaktive Nutzung.""" print("=" * 70) print(" EUROJACKPOT HISTORICAL DATA UPDATER") print("=" * 70) print() # Konfiguration default_data_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv" print("📁 KONFIGURATION") print("=" * 60) print(f"Standard-Datei: {default_data_file}") print() use_default = input("Standard-Datei verwenden? (j/n): ").lower().strip() if use_default in ['j', 'ja', 'y', 'yes', '']: data_file = default_data_file else: data_file = input("Pfad zur Datendatei: ").strip() print() print("🌐 DATENQUELLE WÄHLEN") print("=" * 60) print("1. Auto (versucht alle Quellen)") print("2. eurojackpot.de") print("3. euro-jackpot.net") print("4. Manuelle Eingabe") print() source_choice = input("Wahl (1-4): ").strip() source_map = { '1': 'auto', '2': 'eurojackpot_de', '3': 'euro_jackpot_net', '4': 'manual' } source = source_map.get(source_choice, 'auto') print() print("🧪 DRY RUN?") print("=" * 60) print("Dry Run = Keine Änderungen, nur Vorschau") dry_run_choice = input("Dry Run aktivieren? (j/n): ").lower().strip() dry_run = dry_run_choice in ['j', 'ja', 'y', 'yes'] print() # Update durchführen updater = EurojackpotDataUpdater(data_file) success = updater.update(source=source, dry_run=dry_run) print() print("=" * 70) if success: print("✅ UPDATE ERFOLGREICH ABGESCHLOSSEN") else: print("❌ UPDATE FEHLGESCHLAGEN") print("=" * 70) return success if __name__ == "__main__": import sys # Einfache CLI if len(sys.argv) > 1: # Kommandozeilen-Modus data_file = sys.argv[1] source = sys.argv[2] if len(sys.argv) > 2 else 'auto' dry_run = '--dry-run' in sys.argv updater = EurojackpotDataUpdater(data_file) success = updater.update(source=source, dry_run=dry_run) sys.exit(0 if success else 1) else: # Interaktiver Modus main()