Initial commit: Eurojackpot analysis and prediction system

This repository contains a comprehensive Eurojackpot lottery analysis and prediction system including:
- Historical data analysis and processing
- ML-based prediction models
- Automated weekly tip generation
- Position and range analysis tools
- Notification system for results

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-16 15:53:01 +01:00
co-authored by Claude Sonnet 4.5
commit 70e0638dee
67 changed files with 8491869 additions and 0 deletions
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python3
"""
Erstellt Beispiel-CSV-Dateien für das Eurojackpot-Processing
"""
import pandas as pd
import itertools
from datetime import datetime, timedelta
import random
def create_all_combinations():
"""Erstellt eine CSV mit allen möglichen 5-aus-50 Kombinationen (Beispiel mit ersten 100)."""
print("Erstelle Beispiel-Kombinationsdatei...")
# Für Demo: nur erste 100 Kombinationen von 5 aus 20 Zahlen
# (Vollständige 5-aus-50 Kombinationen wären über 2 Millionen!)
combinations = []
count = 0
for combo in itertools.combinations(range(1, 21), 5): # 5 aus 20 für Demo
combinations.append({
'Zahl1': combo[0],
'Zahl2': combo[1],
'Zahl3': combo[2],
'Zahl4': combo[3],
'Zahl5': combo[4]
})
count += 1
if count >= 100: # Limitierung für Demo
break
df = pd.DataFrame(combinations)
df.to_csv('alle_kombinationen_beispiel.csv', index=False)
print(f"Beispiel-Kombinationsdatei erstellt: {len(df)} Kombinationen")
return df
def create_drawn_numbers():
"""Erstellt eine CSV mit gezogenen Zahlen (Beispieldaten)."""
print("Erstelle Beispiel-Datei mit gezogenen Zahlen...")
drawn_numbers = []
start_date = datetime(2023, 1, 1)
# 20 zufällige Ziehungen erstellen
for i in range(20):
date = start_date + timedelta(days=i*7) # Wöchentliche Ziehungen
# 5 zufällige Zahlen zwischen 1 und 20 (ohne Wiederholung)
numbers = sorted(random.sample(range(1, 21), 5))
drawn_numbers.append({
'Datum': date.strftime('%Y-%m-%d'),
'Z1': numbers[0],
'Z2': numbers[1],
'Z3': numbers[2],
'Z4': numbers[3],
'Z5': numbers[4]
})
df = pd.DataFrame(drawn_numbers)
df.to_csv('gezogene_zahlen_beispiel.csv', index=False)
print(f"Beispiel-Datei mit gezogenen Zahlen erstellt: {len(df)} Ziehungen")
return df
if __name__ == "__main__":
print("=== Erstelle Beispiel-CSV-Dateien ===\n")
# Seed für reproduzierbare Ergebnisse
random.seed(42)
combinations_df = create_all_combinations()
drawn_df = create_drawn_numbers()
print("\n=== Beispieldateien erstellt ===")
print("- alle_kombinationen_beispiel.csv")
print("- gezogene_zahlen_beispiel.csv")
print("\nSie können jetzt das Hauptscript testen:")
+73
View File
@@ -0,0 +1,73 @@
#!/bin/bash
#
# Lädt Eurojackpot-Daten von lotto-datenbank.de
# Quelle: https://lotto-datenbank.de
#
echo "🔄 EUROJACKPOT CSV DOWNLOADER"
echo "=========================================="
echo "Quelle: lotto-datenbank.de"
echo "=========================================="
echo ""
# Ziel-Datei
TARGET_FILE="$HOME/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv"
BACKUP_DIR="$HOME/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/backups"
TEMP_FILE="/tmp/eurojackpot_download_$(date +%Y%m%d_%H%M%S).csv"
# CSV Download URL (aufsteigend sortiert seit 2012)
DOWNLOAD_URL="https://lotto-datenbank.de/download.php?p1=5&p3=0.75&p31=en"
echo "📥 Download von lotto-datenbank.de..."
echo " URL: $DOWNLOAD_URL"
echo ""
# Download mit curl
if curl -L -o "$TEMP_FILE" "$DOWNLOAD_URL" 2>/dev/null; then
echo " ✅ Download erfolgreich"
# Prüfe Dateigröße
FILE_SIZE=$(wc -c < "$TEMP_FILE")
echo " 📊 Dateigröße: $FILE_SIZE bytes"
if [ "$FILE_SIZE" -lt 1000 ]; then
echo " ❌ Datei zu klein - möglicherweise Fehler"
rm "$TEMP_FILE"
exit 1
fi
# Zeige erste Zeilen
echo ""
echo " 📋 Vorschau (erste 5 Zeilen):"
head -5 "$TEMP_FILE" | sed 's/^/ /'
echo ""
# Backup erstellen falls Datei existiert
if [ -f "$TARGET_FILE" ]; then
echo "💾 Erstelle Backup..."
mkdir -p "$BACKUP_DIR"
BACKUP_NAME="AlleEurojackpotzahlen.csv.backup_$(date +%Y%m%d_%H%M%S)"
cp "$TARGET_FILE" "$BACKUP_DIR/$BACKUP_NAME"
echo " ✅ Backup: $BACKUP_NAME"
echo ""
fi
# Verschiebe neue Datei
echo "💾 Speichere neue Datei..."
mv "$TEMP_FILE" "$TARGET_FILE"
# Zähle Ziehungen (Zeilen - Header)
LINE_COUNT=$(($(wc -l < "$TARGET_FILE") - 1))
echo "$LINE_COUNT Ziehungen gespeichert"
echo " 📁 Datei: $(basename "$TARGET_FILE")"
echo ""
echo "=========================================="
echo "✅ UPDATE ERFOLGREICH"
echo "=========================================="
else
echo " ❌ Download fehlgeschlagen"
rm -f "$TEMP_FILE"
exit 1
fi
+165
View File
@@ -0,0 +1,165 @@
#!/usr/bin/env python3
"""
Eurojackpot CSV Processor
Dieses Script liest zwei CSV-Dateien ein:
1. Alle möglichen Zahlenkombinationen (5 Zahlen)
2. Bereits gezogene Zahlen mit Datumsstempel
Es markiert in der ersten Datei alle bereits gezogenen Kombinationen mit 1 (sonst 0).
"""
import pandas as pd
import sys
from pathlib import Path
def load_combinations_file(filepath):
"""Lädt die Datei mit allen möglichen Kombinationen."""
try:
df = pd.read_csv(filepath)
print(f"Kombinationen geladen: {len(df)} Zeilen")
print(f"Spalten: {list(df.columns)}")
return df
except Exception as e:
print(f"Fehler beim Laden der Kombinationsdatei: {e}")
return None
def load_drawn_numbers_file(filepath):
"""Lädt die Datei mit bereits gezogenen Zahlen."""
try:
df = pd.read_csv(filepath)
print(f"Gezogene Zahlen geladen: {len(df)} Zeilen")
print(f"Spalten: {list(df.columns)}")
return df
except Exception as e:
print(f"Fehler beim Laden der gezogenen Zahlen: {e}")
return None
def create_combination_key(row, z_columns):
"""Erstellt einen eindeutigen Schlüssel aus den 5 Zahlen (sortiert)."""
numbers = [row[col] for col in z_columns]
return tuple(sorted(numbers))
def process_eurojackpot_data(combinations_file, drawn_numbers_file, output_file):
"""Hauptfunktion zur Verarbeitung der Eurojackpot-Daten."""
# CSV-Dateien laden
print("Lade Kombinationsdatei...")
combinations_df = load_combinations_file(combinations_file)
if combinations_df is None:
return False
print("\nLade Datei mit gezogenen Zahlen...")
drawn_df = load_drawn_numbers_file(drawn_numbers_file)
if drawn_df is None:
return False
# Spalten für Zahlen identifizieren
# Annahme: In der Kombinationsdatei sind die ersten 5 Spalten die Zahlen
# In der gezogenen Zahlen-Datei sind es Z1, Z2, Z3, Z4, Z5
# Für Kombinationsdatei - erste 5 numerische Spalten verwenden
numeric_cols = combinations_df.select_dtypes(include=['number']).columns
if len(numeric_cols) >= 5:
combo_z_columns = numeric_cols[:5].tolist()
else:
# Fallback: erste 5 Spalten nehmen
combo_z_columns = combinations_df.columns[:5].tolist()
print(f"Verwendete Spalten für Kombinationen: {combo_z_columns}")
# Für gezogene Zahlen - Z1 bis Z5 Spalten suchen
z_columns = [col for col in drawn_df.columns if col.startswith('Z') and col[1:].isdigit()]
z_columns = sorted(z_columns)[:5] # Ersten 5 Z-Spalten nehmen
if not z_columns:
# Fallback: nach Spalten mit "Zahl" im Namen suchen oder numerische Spalten
z_columns = [col for col in drawn_df.columns if 'zahl' in col.lower()][:5]
if not z_columns:
z_columns = drawn_df.select_dtypes(include=['number']).columns[:5].tolist()
print(f"Verwendete Spalten für gezogene Zahlen: {z_columns}")
# Set mit allen gezogenen Kombinationen erstellen
print("\nErstelle Set mit gezogenen Kombinationen...")
drawn_combinations = set()
for _, row in drawn_df.iterrows():
combo_key = create_combination_key(row, z_columns)
drawn_combinations.add(combo_key)
print(f"Anzahl eindeutige gezogene Kombinationen: {len(drawn_combinations)}")
# Neue Spalte für Markierungen hinzufügen
print("\nMarkiere gezogene Kombinationen...")
combinations_df['bereits_gezogen'] = 0
marked_count = 0
for idx, row in combinations_df.iterrows():
combo_key = create_combination_key(row, combo_z_columns)
if combo_key in drawn_combinations:
combinations_df.at[idx, 'bereits_gezogen'] = 1
marked_count += 1
print(f"Anzahl markierte Kombinationen: {marked_count}")
# Ergebnis speichern
print(f"\nSpeichere Ergebnis in: {output_file}")
combinations_df.to_csv(output_file, index=False)
# Statistiken ausgeben
total_combinations = len(combinations_df)
drawn_percentage = (marked_count / total_combinations) * 100 if total_combinations > 0 else 0
print(f"\n=== STATISTIKEN ===")
print(f"Gesamte Kombinationen: {total_combinations:,}")
print(f"Bereits gezogene Kombinationen: {marked_count:,}")
print(f"Prozentsatz bereits gezogen: {drawn_percentage:.4f}%")
print(f"Noch nicht gezogene Kombinationen: {total_combinations - marked_count:,}")
return True
def main():
"""Hauptfunktion mit Benutzerinteraktion."""
print("=== Eurojackpot CSV Processor ===\n")
# Dateipfade abfragen oder Standard verwenden
if len(sys.argv) >= 4:
combinations_file = sys.argv[1]
drawn_numbers_file = sys.argv[2]
output_file = sys.argv[3]
else:
print("Geben Sie die Dateipfade ein (oder drücken Sie Enter für Standard):")
combinations_file = input("Pfad zur Kombinationsdatei (alle_kombinationen.csv): ").strip()
if not combinations_file:
combinations_file = "alle_kombinationen.csv"
drawn_numbers_file = input("Pfad zur Datei mit gezogenen Zahlen (gezogene_zahlen.csv): ").strip()
if not drawn_numbers_file:
drawn_numbers_file = "gezogene_zahlen.csv"
output_file = input("Pfad für Ausgabedatei (kombinationen_markiert.csv): ").strip()
if not output_file:
output_file = "kombinationen_markiert.csv"
# Überprüfen ob Dateien existieren
if not Path(combinations_file).exists():
print(f"Fehler: Kombinationsdatei '{combinations_file}' nicht gefunden!")
return
if not Path(drawn_numbers_file).exists():
print(f"Fehler: Datei mit gezogenen Zahlen '{drawn_numbers_file}' nicht gefunden!")
return
# Verarbeitung starten
success = process_eurojackpot_data(combinations_file, drawn_numbers_file, output_file)
if success:
print(f"\n✅ Verarbeitung erfolgreich abgeschlossen!")
print(f"Ergebnis gespeichert in: {output_file}")
else:
print("\n❌ Fehler bei der Verarbeitung!")
if __name__ == "__main__":
main()
@@ -0,0 +1,220 @@
#!/usr/bin/env python3
"""
Eurojackpot CSV Processor - Angepasst für Semikolon-getrennte CSV-Dateien
Dieses Script liest zwei CSV-Dateien ein:
1. Alle möglichen Zahlenkombinationen (5 Zahlen)
2. Bereits gezogene Zahlen mit Datumsstempel
Es markiert in der ersten Datei alle bereits gezogenen Kombinationen mit 1 (sonst 0).
"""
import pandas as pd
import sys
from pathlib import Path
def detect_separator(filepath):
"""Erkennt das CSV-Trennzeichen automatisch."""
try:
with open(filepath, 'r', encoding='utf-8') as f:
first_line = f.readline()
if ';' in first_line and first_line.count(';') > first_line.count(','):
return ';'
return ','
except:
return ','
def load_combinations_file(filepath):
"""Lädt die Datei mit allen möglichen Kombinationen."""
try:
sep = detect_separator(filepath)
print(f"Erkanntes Trennzeichen für Kombinationen: '{sep}'")
df = pd.read_csv(filepath, sep=sep)
print(f"Kombinationen geladen: {len(df)} Zeilen")
print(f"Spalten: {list(df.columns)}")
return df
except Exception as e:
print(f"Fehler beim Laden der Kombinationsdatei: {e}")
return None
def load_drawn_numbers_file(filepath):
"""Lädt die Datei mit bereits gezogenen Zahlen."""
try:
sep = detect_separator(filepath)
print(f"Erkanntes Trennzeichen für gezogene Zahlen: '{sep}'")
df = pd.read_csv(filepath, sep=sep)
print(f"Gezogene Zahlen geladen: {len(df)} Zeilen")
print(f"Spalten: {list(df.columns)}")
return df
except Exception as e:
print(f"Fehler beim Laden der gezogenen Zahlen: {e}")
return None
def identify_number_columns(df, is_combinations=True):
"""Identifiziert die Spalten mit den Zahlen."""
columns = df.columns.tolist()
if is_combinations:
# Für Kombinationsdatei: Z1, Z2, Z3, Z4, Z5 suchen
z_cols = [col for col in columns if col.startswith('Z') and len(col) == 2 and col[1:].isdigit()]
z_cols = sorted(z_cols)[:5]
if len(z_cols) >= 5:
return z_cols
# Fallback: erste 5 numerische Spalten
numeric_cols = df.select_dtypes(include=['number']).columns[:5].tolist()
if len(numeric_cols) >= 5:
return numeric_cols
# Fallback: erste 5 Spalten
return columns[:5]
else:
# Für gezogene Zahlen: Z1-Z5 suchen (nicht SZ1, SZ2)
z_cols = [col for col in columns if col.startswith('Z') and len(col) == 2 and col[1:].isdigit()]
z_cols = [col for col in z_cols if not col.startswith('SZ')] # Superzahlen ausschließen
z_cols = sorted(z_cols)[:5]
if len(z_cols) >= 5:
return z_cols
# Fallback: numerische Spalten (ohne Datum)
numeric_cols = df.select_dtypes(include=['number']).columns
numeric_cols = [col for col in numeric_cols if 'datum' not in col.lower()][:5]
if len(numeric_cols) >= 5:
return numeric_cols.tolist()
# Letzter Fallback
return columns[:5]
def create_combination_key(row, z_columns):
"""Erstellt einen eindeutigen Schlüssel aus den 5 Zahlen (sortiert)."""
try:
numbers = [int(row[col]) for col in z_columns]
return tuple(sorted(numbers))
except:
return None
def process_eurojackpot_data(combinations_file, drawn_numbers_file, output_file):
"""Hauptfunktion zur Verarbeitung der Eurojackpot-Daten."""
# CSV-Dateien laden
print("Lade Kombinationsdatei...")
combinations_df = load_combinations_file(combinations_file)
if combinations_df is None:
return False
print("\nLade Datei mit gezogenen Zahlen...")
drawn_df = load_drawn_numbers_file(drawn_numbers_file)
if drawn_df is None:
return False
# Spalten für Zahlen identifizieren
combo_z_columns = identify_number_columns(combinations_df, is_combinations=True)
drawn_z_columns = identify_number_columns(drawn_df, is_combinations=False)
print(f"\nVerwendete Spalten für Kombinationen: {combo_z_columns}")
print(f"Verwendete Spalten für gezogene Zahlen: {drawn_z_columns}")
# Datencheck
print(f"\nErste Kombination: {combinations_df[combo_z_columns].iloc[0].tolist()}")
print(f"Erste gezogene Zahlen: {drawn_df[drawn_z_columns].iloc[0].tolist()}")
# Set mit allen gezogenen Kombinationen erstellen
print("\nErstelle Set mit gezogenen Kombinationen...")
drawn_combinations = set()
for _, row in drawn_df.iterrows():
combo_key = create_combination_key(row, drawn_z_columns)
if combo_key:
drawn_combinations.add(combo_key)
print(f"Anzahl eindeutige gezogene Kombinationen: {len(drawn_combinations)}")
# Beispiele anzeigen
if drawn_combinations:
print(f"Erste 5 gezogene Kombinationen: {list(drawn_combinations)[:5]}")
# Neue Spalte für Markierungen hinzufügen (falls noch nicht vorhanden)
if 'bereits_gezogen' not in combinations_df.columns:
combinations_df['bereits_gezogen'] = 0
else:
combinations_df['bereits_gezogen'] = 0 # Zurücksetzen
print("\nMarkiere gezogene Kombinationen...")
marked_count = 0
for idx, row in combinations_df.iterrows():
combo_key = create_combination_key(row, combo_z_columns)
if combo_key and combo_key in drawn_combinations:
combinations_df.at[idx, 'bereits_gezogen'] = 1
marked_count += 1
if marked_count <= 5: # Erste 5 Treffer anzeigen
print(f"Treffer gefunden: {combo_key}")
print(f"Anzahl markierte Kombinationen: {marked_count}")
# Ergebnis speichern
print(f"\nSpeichere Ergebnis in: {output_file}")
sep = detect_separator(combinations_file) # Gleiches Trennzeichen wie Eingabe verwenden
combinations_df.to_csv(output_file, sep=sep, index=False)
# Statistiken ausgeben
total_combinations = len(combinations_df)
drawn_percentage = (marked_count / total_combinations) * 100 if total_combinations > 0 else 0
print(f"\n=== STATISTIKEN ===")
print(f"Gesamte Kombinationen: {total_combinations:,}")
print(f"Bereits gezogene Kombinationen: {marked_count:,}")
print(f"Prozentsatz bereits gezogen: {drawn_percentage:.4f}%")
print(f"Noch nicht gezogene Kombinationen: {total_combinations - marked_count:,}")
return True
def main():
"""Hauptfunktion mit Benutzerinteraktion."""
print("=== Eurojackpot CSV Processor (Fixed) ===\n")
# Dateipfade abfragen oder Standard verwenden
if len(sys.argv) >= 4:
combinations_file = sys.argv[1]
drawn_numbers_file = sys.argv[2]
output_file = sys.argv[3]
else:
print("Geben Sie die Dateipfade ein (oder drücken Sie Enter für Standard):")
combinations_file = input("Pfad zur Kombinationsdatei: ").strip()
if not combinations_file:
combinations_file = "Alle_Eurojackpot_Kombinationen_mit_Status.csv"
drawn_numbers_file = input("Pfad zur Datei mit gezogenen Zahlen: ").strip()
if not drawn_numbers_file:
drawn_numbers_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv"
output_file = input("Pfad für Ausgabedatei: ").strip()
if not output_file:
output_file = "Kombinationen_markiert_fixed.csv"
# Überprüfen ob Dateien existieren
if not Path(combinations_file).exists():
print(f"Fehler: Kombinationsdatei '{combinations_file}' nicht gefunden!")
return
if not Path(drawn_numbers_file).exists():
print(f"Fehler: Datei mit gezogenen Zahlen '{drawn_numbers_file}' nicht gefunden!")
return
# Verarbeitung starten
success = process_eurojackpot_data(combinations_file, drawn_numbers_file, output_file)
if success:
print(f"\n✅ Verarbeitung erfolgreich abgeschlossen!")
print(f"Ergebnis gespeichert in: {output_file}")
else:
print("\n❌ Fehler bei der Verarbeitung!")
if __name__ == "__main__":
main()
+62
View File
@@ -0,0 +1,62 @@
#!/usr/bin/env python3
"""
Vereinfachte Version des Eurojackpot Processors
Für spezifische Spaltenstrukturen
"""
import pandas as pd
def process_eurojackpot_simple(combinations_csv, drawn_numbers_csv, output_csv):
"""
Vereinfachte Verarbeitung mit festen Spaltenstrukturen
Parameter:
- combinations_csv: CSV mit allen Kombinationen (Spalten: Zahl1, Zahl2, Zahl3, Zahl4, Zahl5)
- drawn_numbers_csv: CSV mit gezogenen Zahlen (Spalten: Datum, Z1, Z2, Z3, Z4, Z5)
- output_csv: Ausgabedatei
"""
# Dateien laden
print("Lade Kombinationsdatei...")
combinations = pd.read_csv(combinations_csv)
print("Lade gezogene Zahlen...")
drawn = pd.read_csv(drawn_numbers_csv)
# Gezogene Kombinationen als Set erstellen (sortiert für Vergleich)
drawn_sets = set()
for _, row in drawn.iterrows():
# Annahme: Spalten Z1, Z2, Z3, Z4, Z5 enthalten die Zahlen
numbers = sorted([row['Z1'], row['Z2'], row['Z3'], row['Z4'], row['Z5']])
drawn_sets.add(tuple(numbers))
print(f"Gefundene gezogene Kombinationen: {len(drawn_sets)}")
# Neue Spalte hinzufügen
combinations['bereits_gezogen'] = 0
# Jede Kombination prüfen
marked = 0
for idx, row in combinations.iterrows():
# Annahme: erste 5 Spalten enthalten die Zahlenkombination
combo_cols = combinations.columns[:5]
numbers = sorted([row[col] for col in combo_cols])
if tuple(numbers) in drawn_sets:
combinations.at[idx, 'bereits_gezogen'] = 1
marked += 1
# Ergebnis speichern
combinations.to_csv(output_csv, index=False)
print(f"Verarbeitung abgeschlossen!")
print(f"Markierte Kombinationen: {marked}")
print(f"Ergebnis gespeichert in: {output_csv}")
# Beispielaufruf
if __name__ == "__main__":
process_eurojackpot_simple(
"alle_kombinationen.csv",
"gezogene_zahlen.csv",
"kombinationen_markiert.csv"
)
+308
View File
@@ -0,0 +1,308 @@
#!/usr/bin/env python3
"""
Eurojackpot Notification System
Sendet Benachrichtigungen via:
- Telegram Bot
- E-Mail (SMTP)
Konfiguration via config/notifications.json
"""
import os
import json
import requests
from typing import Optional, Dict, List
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
class EurojackpotNotifier:
"""Benachrichtigungssystem für Eurojackpot-Events."""
def __init__(self, config_file: Optional[str] = None):
if config_file is None:
# Default config path
script_dir = os.path.dirname(os.path.abspath(__file__))
project_dir = os.path.dirname(os.path.dirname(script_dir))
config_file = os.path.join(project_dir, "config", "notifications.json")
self.config_file = config_file
self.config = self._load_config()
def _load_config(self) -> dict:
"""Lädt Konfiguration."""
if not os.path.exists(self.config_file):
return {
"telegram": {"enabled": False},
"email": {"enabled": False}
}
try:
with open(self.config_file, 'r') as f:
return json.load(f)
except Exception as e:
print(f"⚠️ Fehler beim Laden der Notification-Config: {e}")
return {
"telegram": {"enabled": False},
"email": {"enabled": False}
}
def send_tips_generated(self, tips: List[Dict], timestamp: str, best_tip: Dict):
"""
Benachrichtigung: Neue Tipps generiert.
Args:
tips: Liste der generierten Tipps
timestamp: Zeitstempel der Generierung
best_tip: Bester Tipp mit höchster Confidence
"""
# Formatiere Nachricht
main_numbers = best_tip.get('main_numbers', '?')
euro_numbers = best_tip.get('euro_numbers', '?')
confidence = best_tip.get('confidence', 0)
strategy = best_tip.get('strategy', 'UNKNOWN')
subject = f"🎲 {len(tips)} neue Eurojackpot-Tipps generiert!"
message = f"""🎲 NEUE EUROJACKPOT-TIPPS GENERIERT
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
📊 Anzahl Tipps: {len(tips)}
⏰ Zeitpunkt: {timestamp}
🏆 BESTER TIPP (höchste Confidence):
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🔢 Hauptzahlen: {main_numbers}
⭐ Eurozahlen: {euro_numbers}
📈 Strategie: {strategy}
💎 Confidence: {confidence:.2%}
💡 Alle Tipps findest du in der CSV-Datei!
Viel Glück! 🍀
"""
# Sende via Telegram
if self.config.get("telegram", {}).get("enabled"):
self._send_telegram(message)
# Sende via Email
if self.config.get("email", {}).get("enabled"):
self._send_email(subject, message)
def send_draw_results(self, draw: Dict, evaluation: Dict, best_match: Dict):
"""
Benachrichtigung: Ziehung evaluiert.
Args:
draw: Gezogene Zahlen
evaluation: Evaluierungsergebnisse
best_match: Bester Tipp
"""
# Formatiere gezogene Zahlen
main = [draw[f'Z{i}'] for i in range(1, 6)]
euro = [draw['SZ1'], draw['SZ2']]
date = draw.get('date', 'unknown')
# Formatiere besten Tipp
best_main_matches = best_match.get('main_matches', 0)
best_euro_matches = best_match.get('euro_matches', 0)
total_matches = best_main_matches + best_euro_matches
# Bewertungs-Emoji
if total_matches >= 5:
emoji = "🎉🎉🎉"
rating = "FANTASTISCH!"
elif total_matches >= 4:
emoji = "🎊"
rating = "Sehr gut!"
elif total_matches >= 3:
emoji = ""
rating = "Gut!"
elif total_matches >= 2:
emoji = "👍"
rating = "OK"
else:
emoji = ""
rating = "Nächstes Mal besser"
subject = f"🎰 Ziehung vom {date}: {total_matches} Treffer!"
message = f"""🎰 EUROJACKPOT-ZIEHUNG AUSGEWERTET
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
📅 Datum: {date}
🎲 GEZOGENE ZAHLEN:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🔢 Hauptzahlen: {'-'.join(map(str, main))}
⭐ Eurozahlen: {'-'.join(map(str, euro))}
{emoji} DEINE BESTEN TREFFER:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🔢 Hauptzahlen: {best_main_matches} Treffer
⭐ Eurozahlen: {best_euro_matches} Treffer
🎯 Gesamt: {total_matches} Treffer
📊 Bewertung: {rating}
📈 DURCHSCHNITT ALLER TIPPS:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
🔢 Main: {evaluation.get('avg_main', 0):.1f} Treffer
⭐ Euro: {evaluation.get('avg_euro', 0):.1f} Treffer
💡 Details im Performance-Report!
"""
# Sende via Telegram
if self.config.get("telegram", {}).get("enabled"):
self._send_telegram(message)
# Sende via Email
if self.config.get("email", {}).get("enabled"):
self._send_email(subject, message)
def send_error(self, error_msg: str, context: str = ""):
"""
Benachrichtigung: Fehler aufgetreten.
Args:
error_msg: Fehlermeldung
context: Kontext (z.B. "Tipp-Generierung")
"""
subject = f"❌ Eurojackpot Fehler: {context}"
message = f"""❌ FEHLER IM EUROJACKPOT-SYSTEM
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
📍 Kontext: {context}
🔴 Fehler: {error_msg}
💡 Bitte System-Logs prüfen!
"""
# Sende nur via Telegram (Fehler sind dringender)
if self.config.get("telegram", {}).get("enabled"):
self._send_telegram(message)
def _send_telegram(self, message: str):
"""Sendet Nachricht via Telegram Bot."""
try:
telegram_config = self.config.get("telegram", {})
bot_token = telegram_config.get("bot_token")
chat_id = telegram_config.get("chat_id")
if not bot_token or not chat_id:
return
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = {
"chat_id": chat_id,
"text": message,
"disable_web_page_preview": True
}
response = requests.post(url, json=payload, timeout=10)
if response.status_code != 200:
error_detail = response.json() if response.text else {}
print(f"⚠️ Telegram API Error: {error_detail}")
response.raise_for_status()
except Exception as e:
print(f"⚠️ Telegram-Benachrichtigung fehlgeschlagen: {e}")
def _send_email(self, subject: str, message: str):
"""Sendet E-Mail via SMTP."""
try:
email_config = self.config.get("email", {})
smtp_server = email_config.get("smtp_server")
smtp_port = email_config.get("smtp_port", 587)
smtp_user = email_config.get("smtp_user")
smtp_password = email_config.get("smtp_password")
from_email = email_config.get("from_email", smtp_user)
to_email = email_config.get("to_email")
if not all([smtp_server, smtp_user, smtp_password, to_email]):
return
# Erstelle E-Mail
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = from_email
msg["To"] = to_email
# Plain text
text_part = MIMEText(message, "plain", "utf-8")
msg.attach(text_part)
# Sende via SMTP
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(smtp_user, smtp_password)
server.send_message(msg)
except Exception as e:
print(f"⚠️ E-Mail-Benachrichtigung fehlgeschlagen: {e}")
def test_notifications(self):
"""Testet alle konfigurierten Benachrichtigungen."""
print("\n🧪 TESTE BENACHRICHTIGUNGEN")
print("=" * 70)
test_message = """🧪 TEST-BENACHRICHTIGUNG
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Das Eurojackpot-Benachrichtigungssystem funktioniert! ✅
Wenn du diese Nachricht erhältst, ist alles korrekt konfiguriert.
"""
# Test Telegram
if self.config.get("telegram", {}).get("enabled"):
print("\n📱 Teste Telegram...")
self._send_telegram(test_message)
print(" ✅ Telegram-Nachricht gesendet")
# Test Email
if self.config.get("email", {}).get("enabled"):
print("\n📧 Teste E-Mail...")
self._send_email("🧪 Eurojackpot Test", test_message)
print(" ✅ E-Mail gesendet")
print("\n" + "=" * 70)
print("✅ Test abgeschlossen - prüfe deine Nachrichten!")
def main():
"""Test-Funktion."""
import argparse
parser = argparse.ArgumentParser(description="Eurojackpot Notifier Test")
parser.add_argument(
'--test',
action='store_true',
help='Sendet Test-Benachrichtigungen'
)
parser.add_argument(
'--config',
type=str,
help='Pfad zur Config-Datei'
)
args = parser.parse_args()
notifier = EurojackpotNotifier(config_file=args.config)
if args.test:
notifier.test_notifications()
else:
print("Nutze --test um Test-Benachrichtigungen zu senden")
if __name__ == "__main__":
main()
+243
View File
@@ -0,0 +1,243 @@
#!/usr/bin/env python3
"""
Einfacher Eurojackpot Data Updater
Manuelle Eingabe oder CSV-Import von neuen Ziehungen.
Perfekt als Fallback wenn Web-Scraping nicht funktioniert.
"""
import pandas as pd
from datetime import datetime
import shutil
import os
def load_existing_data(data_file):
"""Lädt existierende Daten."""
print("📂 Lade existierende Daten...")
if not os.path.exists(data_file):
print(" ⚠️ Datei existiert nicht - erstelle neue")
return pd.DataFrame(columns=['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2'])
df = pd.read_csv(data_file, sep=';')
df['datum'] = pd.to_datetime(df['datum'], format='%Y-%m-%d', errors='coerce')
print(f"{len(df)} Ziehungen geladen")
if len(df) > 0:
print(f" 📅 Neueste: {df['datum'].max().strftime('%Y-%m-%d')}")
return df
def create_backup(data_file):
"""Erstellt Backup."""
if not os.path.exists(data_file):
return
print("\n💾 Erstelle Backup...")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_dir = os.path.join(os.path.dirname(data_file), "backups")
os.makedirs(backup_dir, exist_ok=True)
backup_file = os.path.join(backup_dir, f"{os.path.basename(data_file)}.backup_{timestamp}")
shutil.copy2(data_file, backup_file)
print(f" ✅ Backup: {os.path.basename(backup_file)}")
def manual_input_mode():
"""Manuelle Eingabe von Ziehungen."""
print("\n⌨️ MANUELLE EINGABE")
print("=" * 70)
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
# Sortiere Zahlen
numbers_sorted = sorted(numbers)
euro_sorted = sorted(euro_numbers)
draws.append({
'datum': date_obj,
'Z1': numbers_sorted[0],
'Z2': numbers_sorted[1],
'Z3': numbers_sorted[2],
'Z4': numbers_sorted[3],
'Z5': numbers_sorted[4],
'SZ1': euro_sorted[0],
'SZ2': euro_sorted[1]
})
print(f" ✅ Hinzugefügt: {date_obj.strftime('%Y-%m-%d')} | "
f"{numbers_sorted[0]}-{numbers_sorted[1]}-{numbers_sorted[2]}-{numbers_sorted[3]}-{numbers_sorted[4]} | "
f"{euro_sorted[0]}-{euro_sorted[1]}")
except ValueError as e:
print(f" ❌ Fehler: {e}")
continue
return draws
def csv_import_mode():
"""CSV-Import von Ziehungen."""
print("\n📁 CSV-IMPORT")
print("=" * 70)
print("CSV-Format: datum;Z1;Z2;Z3;Z4;Z5;SZ1;SZ2")
print("Beispiel: 2025-01-24;7;18;26;37;46;3;11\n")
csv_file = input("Pfad zur CSV-Datei: ").strip()
if not os.path.exists(csv_file):
print(f" ❌ Datei nicht gefunden: {csv_file}")
return []
try:
df_import = pd.read_csv(csv_file, sep=';')
df_import['datum'] = pd.to_datetime(df_import['datum'], format='%Y-%m-%d', errors='coerce')
draws = df_import.to_dict('records')
print(f"{len(draws)} Ziehungen aus CSV geladen")
# Zeige erste 3
for i, draw in enumerate(draws[:3]):
print(f" {i+1}. {draw['datum'].strftime('%Y-%m-%d')} | "
f"{draw['Z1']}-{draw['Z2']}-{draw['Z3']}-{draw['Z4']}-{draw['Z5']} | "
f"{draw['SZ1']}-{draw['SZ2']}")
if len(draws) > 3:
print(f" ... und {len(draws) - 3} weitere")
return draws
except Exception as e:
print(f" ❌ Fehler beim Laden: {e}")
return []
def merge_and_save(df_existing, new_draws, data_file):
"""Merged und speichert Daten."""
if not new_draws:
print("\n⚠️ Keine neuen Ziehungen zum Hinzufügen")
return False
print(f"\n🔀 Merge {len(new_draws)} neue Ziehung(en)...")
df_new = pd.DataFrame(new_draws)
df_combined = pd.concat([df_existing, df_new], ignore_index=True)
# Duplikate entfernen
before = len(df_combined)
df_combined = df_combined.drop_duplicates(subset=['datum'], keep='first')
after = len(df_combined)
if before - after > 0:
print(f" 🗑️ {before - after} Duplikat(e) entfernt")
# Sortieren
df_combined = df_combined.sort_values('datum', ascending=True).reset_index(drop=True)
# Speichern
print(f"\n💾 Speichere {len(df_combined)} Ziehungen...")
df_export = df_combined.copy()
df_export['datum'] = df_export['datum'].dt.strftime('%Y-%m-%d')
df_export.to_csv(data_file, sep=';', index=False)
print(f" ✅ Gespeichert: {data_file}")
print(f" 📊 Vorher: {len(df_existing)} | Neu: {len(df_combined) - len(df_existing)} | Gesamt: {len(df_combined)}")
return True
def main():
"""Hauptfunktion."""
print()
print("=" * 70)
print(" EUROJACKPOT SIMPLE UPDATER")
print(" Manuelle Eingabe oder CSV-Import")
print("=" * 70)
print()
# Datei-Pfad
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv"
print(f"Standard-Datei: {default_file}")
use_default = input("\nStandard verwenden? (j/n): ").strip().lower()
if use_default in ['j', 'ja', 'y', 'yes', '']:
data_file = default_file
else:
data_file = input("Pfad zur Datei: ").strip()
print()
# Lade existierende Daten
df_existing = load_existing_data(data_file)
# Backup
create_backup(data_file)
# Eingabe-Modus wählen
print("\n📝 EINGABE-MODUS")
print("=" * 70)
print("1. Manuelle Eingabe (einzelne Ziehungen)")
print("2. CSV-Import (mehrere Ziehungen)")
print()
mode = input("Wahl (1/2): ").strip()
if mode == '2':
new_draws = csv_import_mode()
else:
new_draws = manual_input_mode()
# Merge und speichern
success = merge_and_save(df_existing, new_draws, data_file)
print()
print("=" * 70)
if success:
print("✅ UPDATE ERFOLGREICH")
else:
print("⚠️ KEINE ÄNDERUNGEN")
print("=" * 70)
if __name__ == "__main__":
main()
+556
View File
@@ -0,0 +1,556 @@
#!/usr/bin/env python3
"""
Eurojackpot API Data Updater
Lädt Eurojackpot-Ziehungen von verschiedenen APIs:
1. lottoAPI (https://lottoapi.herokuapp.com)
2. Lottoland API
3. Sazka.cz API (Fallback)
Features:
- Automatisches Laden von APIs
- Fallback auf andere APIs wenn primäre fehlschlägt
- Duplikate-Vermeidung
- Automatisches Backup
"""
import requests
import pandas as pd
from datetime import datetime
import shutil
import os
from typing import List, Dict, Optional
import time
class EurojackpotAPIUpdater:
"""Updater für Eurojackpot über APIs."""
def __init__(self, data_file: str):
self.data_file = data_file
self.backup_file = None
self.df_existing = None
# API-Endpoints
self.apis = {
'lottoapi': {
'url': 'https://lottoapi.herokuapp.com/eurojackpot-results/100',
'name': 'lottoAPI',
'parser': self._parse_lottoapi
},
'lottoland': {
'url': 'https://media.lottoland.com/api/drawings/euroJackpot',
'name': 'Lottoland API',
'parser': self._parse_lottoland
},
'sazka': {
'url': 'https://www.sazka.cz/api/draw-info/past-draws/eurojackpot',
'name': 'Sazka.cz API',
'parser': self._parse_sazka
}
}
print("🔄 EUROJACKPOT API UPDATER")
print("=" * 70)
print(f"📁 Datei: {os.path.basename(data_file)}")
print("=" * 70)
def load_existing_data(self) -> bool:
"""Lädt existierende Daten."""
print("\n📂 Lade existierende Daten...")
try:
if not os.path.exists(self.data_file):
print(f" ⚠️ Datei existiert nicht - erstelle neue")
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=';')
# Konvertiere Datum
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"{len(self.df_existing)} Ziehungen geladen")
if len(self.df_existing) > 0:
latest = self.df_existing['datum'].max()
oldest = self.df_existing['datum'].min()
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
return True
except Exception as e:
print(f" ❌ Fehler: {e}")
return False
def create_backup(self) -> bool:
"""Erstellt Backup."""
if not os.path.exists(self.data_file):
return True
try:
print("\n💾 Erstelle Backup...")
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: {os.path.basename(self.backup_file)}")
return True
except Exception as e:
print(f" ⚠️ Backup-Fehler: {e}")
return False
def fetch_from_api(self, api_name: str) -> List[Dict]:
"""
Holt Daten von spezifischer API.
Args:
api_name: Name der API ('lottoapi', 'lottoland', 'sazka')
Returns:
Liste von Ziehungen
"""
if api_name not in self.apis:
print(f" ❌ Unbekannte API: {api_name}")
return []
api = self.apis[api_name]
print(f"\n🌐 Versuche {api['name']}...")
print(f" URL: {api['url']}")
try:
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
}
response = requests.get(api['url'], headers=headers, timeout=15)
response.raise_for_status()
print(f" ✅ Antwort erhalten ({len(response.content)} bytes)")
# Parse mit spezifischem Parser
draws = api['parser'](response.json())
if draws:
print(f"{len(draws)} Ziehungen extrahiert")
else:
print(f" ⚠️ Keine Ziehungen extrahiert")
return draws
except requests.RequestException as e:
print(f" ❌ Netzwerkfehler: {e}")
return []
except Exception as e:
print(f" ❌ Fehler: {e}")
return []
def _parse_lottoapi(self, data: List[Dict]) -> List[Dict]:
"""
Parst lottoAPI Response.
Expected format:
[
{
"date": "2025-01-17",
"numbers": [3, 9, 12, 24, 39],
"euroNumbers": [5, 10]
},
...
]
"""
draws = []
try:
for item in data:
# Datum
date_str = item.get('date')
date_obj = datetime.strptime(date_str, '%Y-%m-%d')
# Hauptzahlen
numbers = sorted(item.get('numbers', []))
if len(numbers) != 5:
continue
# Eurozahlen
euro_numbers = sorted(item.get('euroNumbers', []))
if len(euro_numbers) != 2:
continue
draw = {
'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]
}
if self._validate_draw(draw):
draws.append(draw)
except Exception as e:
print(f" ⚠️ Parse-Fehler: {e}")
return draws
def _parse_lottoland(self, data: Dict) -> List[Dict]:
"""
Parst Lottoland API Response.
Expected format: {'last': {...}, 'next': {...}}
"""
draws = []
try:
# Lottoland gibt nur die letzte Ziehung zurück
if 'last' in data:
item = data['last']
# Datum extrahieren aus verschachteltem 'date' Objekt
date_info = item.get('date', {})
day = date_info.get('day')
month = date_info.get('month')
year = date_info.get('year')
if day and month and year:
date_obj = datetime(year, month, day)
else:
return draws
# Hauptzahlen
numbers = sorted(item.get('numbers', []))
if len(numbers) != 5:
return draws
# Eurozahlen
euro_numbers = sorted(item.get('euroNumbers', []))
if len(euro_numbers) != 2:
return draws
draw = {
'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]
}
if self._validate_draw(draw):
draws.append(draw)
print(f" ✅ Ziehung vom {date_obj.strftime('%Y-%m-%d')} extrahiert")
except Exception as e:
print(f" ⚠️ Parse-Fehler: {e}")
return draws
def _parse_sazka(self, data: Dict) -> List[Dict]:
"""
Parst Sazka.cz API Response.
Sazka hat zwei Steps:
1. Liste von Draw IDs holen
2. Für jede Draw ID Details holen
"""
draws = []
try:
# Erste Response enthält Liste von Ziehungen
if isinstance(data, list):
past_draws = data
elif isinstance(data, dict) and 'draws' in data:
past_draws = data['draws']
else:
return draws
print(f" 📊 {len(past_draws)} Ziehungen gefunden, hole Details...")
# Begrenzen auf letzte 10 Ziehungen (um Requests zu sparen)
for i, draw_item in enumerate(past_draws[:10]):
draw_id = draw_item.get('id') # Nicht 'drawId' sondern 'id'!
if not draw_id:
continue
# Hole Details für diese Ziehung
detail_url = f"https://www.sazka.cz/api/draw-info/draws/universal/eurojackpot/{draw_id}"
try:
response = requests.get(detail_url, timeout=10)
detail_data = response.json()
# Parse Detail
date_str = detail_data.get('date') # Nicht 'drawDate' sondern 'date'
date_obj = datetime.strptime(date_str, '%Y-%m-%d')
# Zahlen sind direkt in 'numbers' und 'euroNumbers'
numbers = sorted(detail_data.get('numbers', []))
euro_numbers = sorted(detail_data.get('euroNumbers', []))
if len(numbers) == 5 and len(euro_numbers) == 2:
draw = {
'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]
}
if self._validate_draw(draw):
draws.append(draw)
if (i + 1) % 5 == 0:
print(f"{i + 1}/10 Details geladen...")
time.sleep(0.3) # Rate limiting
except Exception as e:
print(f" ⚠️ Fehler bei Draw ID {draw_id}: {e}")
continue
print(f"{len(draws)} gültige Ziehungen extrahiert")
except Exception as e:
print(f" ⚠️ Parse-Fehler: {e}")
return draws
def _validate_draw(self, draw: Dict) -> bool:
"""Validiert eine Ziehung."""
try:
required_fields = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
if not all(field in draw for field in required_fields):
return False
main_numbers = [draw[f'Z{i}'] for i in range(1, 6)]
if not all(isinstance(n, int) and 1 <= n <= 50 for n in main_numbers):
return False
if len(set(main_numbers)) != 5:
return False
euro_numbers = [draw['SZ1'], draw['SZ2']]
if not all(isinstance(n, int) and 1 <= n <= 12 for n in euro_numbers):
return False
if len(set(euro_numbers)) != 2:
return False
if not isinstance(draw['datum'], datetime):
return False
return True
except Exception:
return False
def fetch_from_all_apis(self) -> List[Dict]:
"""
Versucht alle APIs nacheinander.
Returns:
Liste von Ziehungen
"""
print("\n🔍 SUCHE NACH DATEN VON APIs")
print("=" * 70)
all_draws = []
# Versuche alle APIs
for api_name in ['lottoapi', 'lottoland', 'sazka']:
draws = self.fetch_from_api(api_name)
if draws:
all_draws.extend(draws)
print(f"{len(draws)} Ziehungen von {self.apis[api_name]['name']}")
break # Erste erfolgreiche API nutzen
else:
print(f" ⏭️ Weiter zur nächsten API...")
time.sleep(1) # Pause zwischen APIs
if not all_draws:
print("\n ❌ Keine Daten von APIs erhalten")
return all_draws
def merge_with_existing(self, new_draws: List[Dict]) -> pd.DataFrame:
"""Merged neue Ziehungen mit existierenden Daten."""
print(f"\n🔀 Merge mit existierenden Daten...")
if not new_draws:
print(" ⚠️ Keine neuen Ziehungen zum Mergen")
return self.df_existing
df_new = pd.DataFrame(new_draws)
if self.df_existing is None or len(self.df_existing) == 0:
df_combined = df_new
print(f" ✅ Neue Datei erstellt mit {len(df_combined)} Ziehungen")
else:
df_combined = pd.concat([self.df_existing, df_new], ignore_index=True)
before_dedup = len(df_combined)
df_combined = df_combined.drop_duplicates(subset=['datum'], keep='first')
after_dedup = len(df_combined)
duplicates = before_dedup - after_dedup
if duplicates > 0:
print(f" 🗑️ {duplicates} Duplikat(e) entfernt")
df_combined = df_combined.sort_values('datum', ascending=True)
df_combined = df_combined.reset_index(drop=True)
new_entries = len(df_combined) - (len(self.df_existing) if self.df_existing is not None else 0)
print(f" ✅ Merge abgeschlossen:")
print(f" Vorher: {len(self.df_existing) if self.df_existing is not None else 0} Ziehungen")
print(f" Neue: {new_entries} Ziehungen")
print(f" Gesamt: {len(df_combined)} Ziehungen")
return df_combined
def save_data(self, df: pd.DataFrame) -> bool:
"""Speichert Daten."""
try:
print(f"\n💾 Speichere Daten...")
df_export = df.copy()
df_export['datum'] = df_export['datum'].dt.strftime('%Y-%m-%d')
column_order = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
df_export = df_export[column_order]
df_export.to_csv(self.data_file, sep=';', index=False)
print(f" ✅ Gespeichert: {self.data_file}")
print(f" 📊 {len(df)} Ziehungen total")
if len(df) > 0:
latest = df['datum'].max()
oldest = df['datum'].min()
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
return True
except Exception as e:
print(f" ❌ Speicherfehler: {e}")
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, api_name: str = 'all', create_backup: bool = True) -> bool:
"""
Führt komplettes Update durch.
Args:
api_name: 'all', 'lottoapi', 'lottoland', 'sazka'
create_backup: Backup erstellen
Returns:
True bei Erfolg
"""
print("\n🚀 STARTE UPDATE")
print("=" * 70)
if not self.load_existing_data():
return False
if create_backup:
self.create_backup()
if api_name == 'all':
new_draws = self.fetch_from_all_apis()
else:
new_draws = self.fetch_from_api(api_name)
if not new_draws:
print("\n❌ Keine Daten von APIs geladen")
return False
df_updated = self.merge_with_existing(new_draws)
success = self.save_data(df_updated)
print("\n" + "=" * 70)
if success:
print("✅ UPDATE ERFOLGREICH ABGESCHLOSSEN")
else:
print("❌ UPDATE FEHLGESCHLAGEN")
print("=" * 70)
return success
def main():
"""Hauptfunktion."""
import sys
print()
print("=" * 70)
print(" EUROJACKPOT API UPDATER")
print(" Unterstützt: lottoAPI, Lottoland, Sazka.cz")
print("=" * 70)
print()
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv"
if len(sys.argv) > 1:
data_file = sys.argv[1]
api_name = sys.argv[2] if len(sys.argv) > 2 else 'all'
else:
print(f"Standard-Datei: {default_file}")
use_default = input("\nStandard verwenden? (j/n): ").strip().lower()
if use_default in ['j', 'ja', 'y', 'yes', '']:
data_file = default_file
else:
data_file = input("Pfad zur Datei: ").strip()
print("\n🌐 API WÄHLEN")
print("=" * 60)
print("1. Alle APIs versuchen (empfohlen)")
print("2. lottoAPI")
print("3. Lottoland API")
print("4. Sazka.cz API")
print()
api_choice = input("Wahl (1-4): ").strip()
api_map = {'1': 'all', '2': 'lottoapi', '3': 'lottoland', '4': 'sazka'}
api_name = api_map.get(api_choice, 'all')
print()
updater = EurojackpotAPIUpdater(data_file)
success = updater.update(api_name=api_name)
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
@@ -0,0 +1,449 @@
#!/usr/bin/env python3
"""
Eurojackpot Data Updater für eurojackpot-zahlen.eu
Optimierter Scraper für https://www.eurojackpot-zahlen.eu/eurojackpot-zahlenarchiv.php
Features:
- Automatisches Scraping aller historischen Ziehungen
- Duplikate-Vermeidung
- Automatisches Backup vor Update
- Validierung der Daten
- Fortschrittsanzeige
"""
import pandas as pd
import requests
from bs4 import BeautifulSoup
from datetime import datetime
import shutil
import os
from typing import List, Dict, Optional
class EurojackpotUpdater:
"""Updater für eurojackpot-zahlen.eu"""
def __init__(self, data_file: str):
self.data_file = data_file
self.url = "https://www.eurojackpot-zahlen.eu/eurojackpot-zahlenarchiv.php"
self.backup_file = None
self.df_existing = None
print("🔄 EUROJACKPOT DATA UPDATER")
print("=" * 70)
print(f"📁 Datei: {os.path.basename(data_file)}")
print(f"🌐 Quelle: {self.url}")
print("=" * 70)
def load_existing_data(self) -> bool:
"""Lädt existierende Daten."""
print("\n📂 Lade existierende Daten...")
try:
if not os.path.exists(self.data_file):
print(f" ⚠️ Datei existiert nicht - erstelle neue")
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=';')
# Konvertiere Datum
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"{len(self.df_existing)} Ziehungen geladen")
if len(self.df_existing) > 0:
latest = self.df_existing['datum'].max()
oldest = self.df_existing['datum'].min()
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
return True
except Exception as e:
print(f" ❌ Fehler: {e}")
return False
def create_backup(self) -> bool:
"""Erstellt Backup."""
if not os.path.exists(self.data_file):
return True
try:
print("\n💾 Erstelle Backup...")
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: {os.path.basename(self.backup_file)}")
return True
except Exception as e:
print(f" ⚠️ Backup-Fehler: {e}")
return False
def fetch_draws_from_web(self) -> List[Dict]:
"""
Scrapt alle Ziehungen von eurojackpot-zahlen.eu
Returns:
Liste von Ziehungen
"""
print(f"\n🌐 Lade Daten von {self.url}...")
try:
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
response = requests.get(self.url, headers=headers, timeout=15)
response.raise_for_status()
print(" ✅ Seite erfolgreich geladen")
print(" 🔍 Parse HTML-Struktur...")
soup = BeautifulSoup(response.content, 'html.parser')
# Neue Struktur: divs mit Klasse "zahlen_rahmen"
# Jeder zahlen_rahmen enthält: zahlen_hinweis01 (#), zahlen_hinweis02 (Datum),
# zahlen_hinweis03 (Zahlen), zahlen_hinweis04 (Eurozahlen)
containers = soup.find_all('div', class_='zahlen_rahmen')
if not containers:
print(" ❌ Keine zahlen_rahmen Container gefunden")
# Debug-Info
print(f" 🔍 Debug: Seite enthält {len(soup.find_all('div'))} divs")
print(f" 🔍 Debug: Erste 500 Zeichen: {response.text[:500]}")
return []
print(f" 📊 {len(containers)} Container gefunden")
print(" ⏳ Extrahiere Ziehungen...")
draws = []
# Überspringe ersten Container (Header)
for i, container in enumerate(containers[1:], 1):
try:
# Extrahiere Datum aus <time> Element
time_element = container.find('time', class_='zahlenarchiv_datum')
if not time_element:
continue
date_text = time_element.text.strip()
date_obj = self._parse_date(date_text)
# Extrahiere Hauptzahlen (zahlenarchiv_zahl)
main_number_divs = container.find_all('div', class_='zahlenarchiv_zahl')
main_numbers = [int(div.text.strip()) for div in main_number_divs if div.text.strip().isdigit()]
# Extrahiere Eurozahlen (zahlenarchiv_zz)
euro_number_divs = container.find_all('div', class_='zahlenarchiv_zz')
euro_numbers = [int(div.text.strip()) for div in euro_number_divs if div.text.strip().isdigit()]
# Validierung
if len(main_numbers) != 5:
print(f" ⚠️ Zeile {i+1}: Ungültige Anzahl Hauptzahlen ({len(main_numbers)})")
continue
if len(euro_numbers) != 2:
print(f" ⚠️ Zeile {i+1}: Ungültige Anzahl Eurozahlen ({len(euro_numbers)})")
continue
# Sortiere Hauptzahlen
main_numbers_sorted = sorted(main_numbers)
# Sortiere Eurozahlen
euro_numbers_sorted = sorted(euro_numbers)
draw = {
'datum': date_obj,
'Z1': main_numbers_sorted[0],
'Z2': main_numbers_sorted[1],
'Z3': main_numbers_sorted[2],
'Z4': main_numbers_sorted[3],
'Z5': main_numbers_sorted[4],
'SZ1': euro_numbers_sorted[0],
'SZ2': euro_numbers_sorted[1]
}
# Validiere
if self._validate_draw(draw):
draws.append(draw)
else:
print(f" ⚠️ Zeile {i+1}: Validierung fehlgeschlagen")
# Fortschrittsanzeige
if (i + 1) % 50 == 0:
print(f"{i + 1}/{len(data_rows)} Zeilen verarbeitet...")
except Exception as e:
print(f" ⚠️ Zeile {i+1}: Fehler beim Parsen - {e}")
continue
print(f"{len(draws)} gültige Ziehungen extrahiert")
return draws
except requests.RequestException as e:
print(f" ❌ Netzwerkfehler: {e}")
return []
except Exception as e:
print(f" ❌ Fehler: {e}")
return []
def _parse_date(self, date_str: str) -> datetime:
"""
Parst Datum im Format "DD.MM.YYYY"
Args:
date_str: Datum als String
Returns:
datetime Objekt
"""
try:
# Format: "03.01.2025"
return datetime.strptime(date_str, '%d.%m.%Y')
except ValueError:
try:
# Fallback: "YYYY-MM-DD"
return datetime.strptime(date_str, '%Y-%m-%d')
except ValueError:
# Letzter Fallback: Aktuelles Datum
print(f" ⚠️ Konnte Datum nicht parsen: {date_str}")
return datetime.now()
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(isinstance(n, int) and 1 <= n <= 50 for n in main_numbers):
return False
# Prüfe Eindeutigkeit
if len(set(main_numbers)) != 5:
return False
# Prüfe Eurozahlen (1-12)
euro_numbers = [draw['SZ1'], draw['SZ2']]
if not all(isinstance(n, int) and 1 <= n <= 12 for n in euro_numbers):
return False
# Prüfe Eindeutigkeit
if len(set(euro_numbers)) != 2:
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...")
if not new_draws:
print(" ⚠️ Keine neuen Ziehungen zum Mergen")
return self.df_existing
# Erstelle DataFrame aus neuen Ziehungen
df_new = pd.DataFrame(new_draws)
# Kombiniere
if self.df_existing is None or len(self.df_existing) == 0:
df_combined = df_new
print(f" ✅ Neue Datei erstellt mit {len(df_combined)} Ziehungen")
else:
df_combined = pd.concat([self.df_existing, df_new], ignore_index=True)
# Entferne Duplikate (basierend auf Datum)
before_dedup = len(df_combined)
df_combined = df_combined.drop_duplicates(subset=['datum'], keep='first')
after_dedup = len(df_combined)
duplicates = before_dedup - after_dedup
if duplicates > 0:
print(f" 🗑️ {duplicates} Duplikat(e) entfernt")
# Sortiere nach Datum (aufsteigend)
df_combined = df_combined.sort_values('datum', ascending=True)
df_combined = df_combined.reset_index(drop=True)
# Statistiken
new_entries = len(df_combined) - (len(self.df_existing) if self.df_existing is not None else 0)
print(f" ✅ Merge abgeschlossen:")
print(f" Vorher: {len(self.df_existing) if self.df_existing is not None else 0} Ziehungen")
print(f" Neue: {new_entries} Ziehungen")
print(f" Gesamt: {len(df_combined)} Ziehungen")
return df_combined
def save_data(self, df: pd.DataFrame) -> bool:
"""
Speichert Daten in CSV.
Args:
df: DataFrame zum Speichern
Returns:
True bei Erfolg
"""
try:
print(f"\n💾 Speichere Daten...")
# Erstelle Kopie für Export
df_export = df.copy()
# Format Datum als String
df_export['datum'] = df_export['datum'].dt.strftime('%Y-%m-%d')
# Sortiere Spalten
column_order = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
df_export = df_export[column_order]
# Speichere
df_export.to_csv(self.data_file, sep=';', index=False)
print(f" ✅ Gespeichert: {self.data_file}")
print(f" 📊 {len(df)} Ziehungen total")
if len(df) > 0:
latest = df['datum'].max()
oldest = df['datum'].min()
print(f" 📅 Zeitraum: {oldest.strftime('%Y-%m-%d')} bis {latest.strftime('%Y-%m-%d')}")
return True
except Exception as e:
print(f" ❌ Speicherfehler: {e}")
# Restore backup falls vorhanden
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, create_backup: bool = True) -> bool:
"""
Führt komplettes Update durch.
Args:
create_backup: Backup vor Update erstellen
Returns:
True bei Erfolg
"""
print("\n🚀 STARTE UPDATE")
print("=" * 70)
# 1. Lade existierende Daten
if not self.load_existing_data():
return False
# 2. Backup erstellen
if create_backup:
self.create_backup()
# 3. Hole Ziehungen von Web
new_draws = self.fetch_draws_from_web()
if not new_draws:
print("\n❌ Keine Daten von Website geladen")
return False
# 4. Merge
df_updated = self.merge_with_existing(new_draws)
# 5. Speichern
success = self.save_data(df_updated)
print("\n" + "=" * 70)
if success:
print("✅ UPDATE ERFOLGREICH ABGESCHLOSSEN")
else:
print("❌ UPDATE FEHLGESCHLAGEN")
print("=" * 70)
return success
def main():
"""Hauptfunktion."""
import sys
print()
print("=" * 70)
print(" EUROJACKPOT DATA UPDATER")
print(" Quelle: eurojackpot-zahlen.eu")
print("=" * 70)
print()
# Standard-Pfad
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv"
# Command-line Argumente
if len(sys.argv) > 1:
data_file = sys.argv[1]
else:
print(f"Standard-Datei: {default_file}")
print()
use_default = input("Standard verwenden? (j/n): ").strip().lower()
if use_default in ['j', 'ja', 'y', 'yes', '']:
data_file = default_file
else:
data_file = input("Pfad zur Datei: ").strip()
# Update durchführen
print()
updater = EurojackpotUpdater(data_file)
success = updater.update()
# Exit-Code
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
+681
View File
@@ -0,0 +1,681 @@
#!/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()
+460
View File
@@ -0,0 +1,460 @@
#!/usr/bin/env python3
"""
CSV-Validator für Lotto 6aus49 und Eurojackpot
Validiert CSV-Dateien auf:
- Korrekte Spaltenstruktur
- Datumformat und -konsistenz
- Zahlenbereich und Eindeutigkeit
- Wochentag (nur Mi/Sa für Lotto, nur Di/Fr für Eurojackpot)
- Chronologische Sortierung
- Duplikate
- Fehlende Werte
- Zukunftsdaten
"""
import pandas as pd
from datetime import datetime
import sys
import os
from typing import List, Dict, Tuple
class LottoCSVValidator:
"""Validator für Lotto 6aus49 CSV-Dateien."""
def __init__(self, csv_file: str):
self.csv_file = csv_file
self.df = None
self.errors = []
self.warnings = []
self.lottery_type = self._detect_lottery_type()
def _detect_lottery_type(self) -> str:
"""Erkennt ob Lotto oder Eurojackpot basierend auf Dateiname."""
basename = os.path.basename(self.csv_file).lower()
if 'eurojackpot' in basename:
return 'eurojackpot'
elif 'lotto' in basename:
return 'lotto'
else:
# Versuche anhand der Spalten zu erkennen
return 'unknown'
def load_csv(self) -> bool:
"""Lädt CSV-Datei."""
try:
if not os.path.exists(self.csv_file):
self.errors.append(f"❌ Datei existiert nicht: {self.csv_file}")
return False
self.df = pd.read_csv(self.csv_file, sep=';')
# Auto-detect wenn noch unknown
if self.lottery_type == 'unknown':
if 'SZ2' in self.df.columns:
self.lottery_type = 'eurojackpot'
elif 'SZ' in self.df.columns:
self.lottery_type = 'lotto'
else:
self.errors.append("❌ Kann Lottery-Typ nicht erkennen")
return False
return True
except Exception as e:
self.errors.append(f"❌ Fehler beim Laden: {e}")
return False
def validate_structure(self) -> bool:
"""Validiert Spaltenstruktur."""
if self.lottery_type == 'lotto':
expected_cols = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
elif self.lottery_type == 'eurojackpot':
expected_cols = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
else:
self.errors.append("❌ Unbekannter Lottery-Typ")
return False
actual_cols = list(self.df.columns)
if actual_cols != expected_cols:
self.errors.append(f"❌ Spaltenstruktur falsch")
self.errors.append(f" Erwartet: {expected_cols}")
self.errors.append(f" Gefunden: {actual_cols}")
return False
return True
def validate_dates(self) -> bool:
"""Validiert Datumsspalte."""
valid = True
# Prüfe auf leere Daten
empty_dates = self.df[self.df['datum'].isna() | (self.df['datum'] == '')]
if len(empty_dates) > 0:
self.errors.append(f"{len(empty_dates)} Zeile(n) mit fehlendem Datum:")
for idx in empty_dates.index[:5]: # Zeige max 5
self.errors.append(f" Zeile {idx + 2}")
if len(empty_dates) > 5:
self.errors.append(f" ... und {len(empty_dates) - 5} weitere")
valid = False
# Konvertiere Datum
try:
self.df['datum'] = pd.to_datetime(self.df['datum'], format='%Y-%m-%d', errors='coerce')
except Exception as e:
self.errors.append(f"❌ Fehler beim Parsen der Datumsangaben: {e}")
return False
# Prüfe auf ungültige Datumsangaben
invalid_dates = self.df[self.df['datum'].isna()]
if len(invalid_dates) > 0:
self.errors.append(f"{len(invalid_dates)} ungültige Datumsangaben")
valid = False
# Prüfe auf Zukunftsdaten
today = datetime.now()
future_dates = self.df[self.df['datum'] > today]
if len(future_dates) > 0:
self.errors.append(f"{len(future_dates)} Datum/Daten in der Zukunft:")
for idx, row in future_dates.iterrows():
self.errors.append(f" Zeile {idx + 2}: {row['datum'].strftime('%Y-%m-%d')}")
valid = False
# Prüfe Wochentage
if self.lottery_type == 'lotto':
# Historisch:
# 1955-1965: Sonntags (und manchmal Montags bei Feiertagen)
# 1965-2000: Samstags
# Ab 2000: Mittwoch + Samstag
weekday_names = {0: 'Mo', 1: 'Di', 2: 'Mi', 3: 'Do', 4: 'Fr', 5: 'Sa', 6: 'So'}
else: # eurojackpot
# Eurojackpot: Dienstag + Freitag
weekday_names = {0: 'Mo', 1: 'Di', 2: 'Mi', 3: 'Do', 4: 'Fr', 5: 'Sa', 6: 'So'}
wrong_weekdays = []
for idx, row in self.df.iterrows():
if pd.notna(row['datum']):
weekday = row['datum'].weekday()
date = row['datum']
if self.lottery_type == 'lotto':
# Lotto: Historische Regeln
if date < datetime(1965, 9, 1):
# Bis Aug 1965: Sonntag (und Montag bei Feiertagen)
if weekday not in [0, 6]: # Mo, So
wrong_weekdays.append((idx, date))
elif date < datetime(2000, 12, 1):
# Sep 1965 - Nov 2000: Nur Samstag
if weekday != 5: # Sa
wrong_weekdays.append((idx, date))
else:
# Ab Dez 2000: Mittwoch + Samstag
if weekday not in [2, 5]: # Mi, Sa
wrong_weekdays.append((idx, date))
else:
# Eurojackpot: Dienstag + Freitag
if weekday not in [1, 4]: # Di, Fr
wrong_weekdays.append((idx, date))
if len(wrong_weekdays) > 0:
# Filtere mögliche Feiertags-Sonderziehungen (< 5 pro Zeitraum = OK)
if len(wrong_weekdays) <= 5:
self.warnings.append(f"⚠️ {len(wrong_weekdays)} Ziehung(en) am ungewöhnlichen Wochentag (möglicherweise Feiertage):")
for idx, date in wrong_weekdays:
weekday_name = weekday_names[date.weekday()]
self.warnings.append(f" Zeile {idx + 2}: {date.strftime('%Y-%m-%d')} ({weekday_name})")
else:
self.errors.append(f"{len(wrong_weekdays)} Ziehung(en) am falschen Wochentag:")
for idx, date in wrong_weekdays[:5]:
weekday_name = weekday_names[date.weekday()]
self.errors.append(f" Zeile {idx + 2}: {date.strftime('%Y-%m-%d')} ({weekday_name})")
if len(wrong_weekdays) > 5:
self.errors.append(f" ... und {len(wrong_weekdays) - 5} weitere")
valid = False
return valid
def validate_numbers(self) -> bool:
"""Validiert Zahlenwerte."""
valid = True
if self.lottery_type == 'lotto':
main_cols = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6']
main_range = (1, 49)
sz_cols = ['SZ']
sz_range = (0, 9)
else: # eurojackpot
main_cols = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5']
main_range = (1, 50)
sz_cols = ['SZ1', 'SZ2']
sz_range = (1, 12)
# Prüfe Hauptzahlen
for col in main_cols:
# Fehlende Werte
missing = self.df[self.df[col].isna()]
if len(missing) > 0:
self.errors.append(f"{len(missing)} fehlende Werte in Spalte {col}")
valid = False
# Zahlenbereich
out_of_range = self.df[
(self.df[col] < main_range[0]) |
(self.df[col] > main_range[1])
]
if len(out_of_range) > 0:
self.errors.append(f"{len(out_of_range)} Werte außerhalb {main_range} in {col}")
for idx, row in out_of_range.head(3).iterrows():
self.errors.append(f" Zeile {idx + 2}: {row[col]}")
valid = False
# Prüfe Superzahl(en)
for col in sz_cols:
missing = self.df[self.df[col].isna()]
if len(missing) > 0:
self.warnings.append(f"⚠️ {len(missing)} fehlende Werte in Spalte {col}")
out_of_range = self.df[
(self.df[col] < sz_range[0]) |
(self.df[col] > sz_range[1])
]
if len(out_of_range) > 0:
self.errors.append(f"{len(out_of_range)} Werte außerhalb {sz_range} in {col}")
valid = False
# Prüfe Eindeutigkeit der Hauptzahlen pro Zeile
duplicate_numbers = []
for idx, row in self.df.iterrows():
main_numbers = [row[col] for col in main_cols if pd.notna(row[col])]
if len(main_numbers) != len(set(main_numbers)):
duplicate_numbers.append((idx, main_numbers))
if len(duplicate_numbers) > 0:
self.errors.append(f"{len(duplicate_numbers)} Zeile(n) mit doppelten Hauptzahlen:")
for idx, numbers in duplicate_numbers[:5]:
self.errors.append(f" Zeile {idx + 2}: {numbers}")
if len(duplicate_numbers) > 5:
self.errors.append(f" ... und {len(duplicate_numbers) - 5} weitere")
valid = False
# Prüfe Sortierung der Hauptzahlen pro Zeile
unsorted_rows = []
for idx, row in self.df.iterrows():
main_numbers = [row[col] for col in main_cols if pd.notna(row[col])]
if main_numbers != sorted(main_numbers):
unsorted_rows.append((idx, main_numbers))
if len(unsorted_rows) > 0:
self.warnings.append(f"⚠️ {len(unsorted_rows)} Zeile(n) mit unsortierten Zahlen:")
for idx, numbers in unsorted_rows[:3]:
self.warnings.append(f" Zeile {idx + 2}: {numbers}")
# Prüfe Sortierung der Eurozahlen (nur Eurojackpot)
if self.lottery_type == 'eurojackpot':
unsorted_euro = []
for idx, row in self.df.iterrows():
if pd.notna(row['SZ1']) and pd.notna(row['SZ2']):
if row['SZ1'] > row['SZ2']:
unsorted_euro.append((idx, row['SZ1'], row['SZ2']))
if len(unsorted_euro) > 0:
self.warnings.append(f"⚠️ {len(unsorted_euro)} Zeile(n) mit unsortierten Eurozahlen")
return valid
def validate_duplicates(self) -> bool:
"""Prüft auf doppelte Datumssätze."""
duplicates = self.df[self.df.duplicated(subset=['datum'], keep=False)]
if len(duplicates) > 0:
self.errors.append(f"{len(duplicates)} doppelte Datumseinträge gefunden:")
for idx, row in duplicates.head(5).iterrows():
self.errors.append(f" Zeile {idx + 2}: {row['datum'].strftime('%Y-%m-%d')}")
return False
return True
def validate_chronology(self) -> bool:
"""Prüft chronologische Sortierung."""
if len(self.df) < 2:
return True
unsorted = []
for i in range(len(self.df) - 1):
if self.df.iloc[i]['datum'] > self.df.iloc[i + 1]['datum']:
unsorted.append((i, self.df.iloc[i]['datum'], self.df.iloc[i + 1]['datum']))
if len(unsorted) > 0:
self.warnings.append(f"⚠️ {len(unsorted)} Stelle(n) mit falscher chronologischer Reihenfolge:")
for idx, date1, date2 in unsorted[:3]:
self.warnings.append(f" Zeile {idx + 2}: {date1.strftime('%Y-%m-%d')} > {date2.strftime('%Y-%m-%d')}")
return len(unsorted) == 0
def validate_completeness(self) -> bool:
"""Prüft auf Lücken in den Ziehungen."""
if len(self.df) < 2:
return True
if self.lottery_type == 'lotto':
# Lotto: 2x pro Woche (Mi, Sa)
expected_days = 3.5 # Durchschnitt
else:
# Eurojackpot: 2x pro Woche (Di, Fr)
expected_days = 3.5
gaps = []
for i in range(len(self.df) - 1):
date1 = self.df.iloc[i]['datum']
date2 = self.df.iloc[i + 1]['datum']
diff = (date2 - date1).days
# Wenn Lücke größer als 7 Tage, könnte Ziehung fehlen
if diff > 7:
gaps.append((i, date1, date2, diff))
if len(gaps) > 0:
self.warnings.append(f"⚠️ {len(gaps)} mögliche Lücke(n) in den Ziehungen (>7 Tage):")
for idx, date1, date2, diff in gaps[:5]:
self.warnings.append(f" Zeile {idx + 2}: {date1.strftime('%Y-%m-%d')}{date2.strftime('%Y-%m-%d')} ({diff} Tage)")
if len(gaps) > 5:
self.warnings.append(f" ... und {len(gaps) - 5} weitere")
return True
def get_statistics(self) -> Dict:
"""Erstellt Statistiken."""
if self.df is None or len(self.df) == 0:
return {}
stats = {
'total_draws': len(self.df),
'date_range': (
self.df['datum'].min().strftime('%Y-%m-%d'),
self.df['datum'].max().strftime('%Y-%m-%d')
),
'years_covered': (self.df['datum'].max().year - self.df['datum'].min().year) + 1,
}
return stats
def validate_all(self) -> bool:
"""Führt alle Validierungen durch."""
print(f"\n{'='*70}")
print(f" CSV VALIDATOR")
print(f" Typ: {self.lottery_type.upper()}")
print(f"{'='*70}")
print(f"\n📁 Datei: {os.path.basename(self.csv_file)}")
if not self.load_csv():
return False
print(f"✅ Datei geladen: {len(self.df)} Zeilen")
# Alle Validierungen
checks = [
("Spaltenstruktur", self.validate_structure),
("Datumsangaben", self.validate_dates),
("Zahlenwerte", self.validate_numbers),
("Duplikate", self.validate_duplicates),
("Chronologie", self.validate_chronology),
("Vollständigkeit", self.validate_completeness),
]
print(f"\n🔍 VALIDIERUNG")
print("="*70)
all_valid = True
for check_name, check_func in checks:
try:
result = check_func()
status = "" if result else ""
print(f"{status} {check_name}")
if not result:
all_valid = False
except Exception as e:
print(f"{check_name} - Fehler: {e}")
all_valid = False
# Statistiken
stats = self.get_statistics()
if stats:
print(f"\n📊 STATISTIKEN")
print("="*70)
print(f"Ziehungen gesamt: {stats['total_draws']}")
print(f"Zeitraum: {stats['date_range'][0]} bis {stats['date_range'][1]}")
print(f"Jahre: {stats['years_covered']}")
# Fehler ausgeben
if self.errors:
print(f"\n❌ FEHLER ({len(self.errors)})")
print("="*70)
for error in self.errors:
print(error)
# Warnungen ausgeben
if self.warnings:
print(f"\n⚠️ WARNUNGEN ({len(self.warnings)})")
print("="*70)
for warning in self.warnings:
print(warning)
# Zusammenfassung
print(f"\n{'='*70}")
if all_valid and not self.errors:
print("✅ VALIDIERUNG ERFOLGREICH - Keine Fehler gefunden!")
elif not self.errors and self.warnings:
print("✅ VALIDIERUNG OK - Nur Warnungen (keine kritischen Fehler)")
else:
print("❌ VALIDIERUNG FEHLGESCHLAGEN")
print("="*70)
return all_valid and len(self.errors) == 0
def main():
"""Hauptfunktion."""
import sys
# Standard-Dateien
default_files = {
'lotto': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/AlleLottozahlen.csv',
'eurojackpot': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv'
}
if len(sys.argv) > 1:
# Benutzerdefinierte Datei
csv_file = sys.argv[1]
validator = LottoCSVValidator(csv_file)
success = validator.validate_all()
sys.exit(0 if success else 1)
else:
# Validiere beide Standard-Dateien
print("\n🎲 Validiere beide Lottery-Dateien...\n")
results = {}
for lottery_type, csv_file in default_files.items():
if os.path.exists(csv_file):
validator = LottoCSVValidator(csv_file)
results[lottery_type] = validator.validate_all()
else:
print(f"\n⚠️ {lottery_type.upper()}: Datei nicht gefunden: {csv_file}")
results[lottery_type] = False
# Gesamtergebnis
print("\n" + "="*70)
print(" GESAMTERGEBNIS")
print("="*70)
for lottery_type, success in results.items():
status = "" if success else ""
print(f"{status} {lottery_type.upper()}")
print("="*70)
all_success = all(results.values())
sys.exit(0 if all_success else 1)
if __name__ == "__main__":
main()
+401
View File
@@ -0,0 +1,401 @@
#!/usr/bin/env python3
"""
Ziehungs-Verifizierer für Lotto 6aus49 und Eurojackpot
Verifiziert Ziehungen gegen offizielle Datenquellen:
1. GitHub Lotto Archive (für Lotto 6aus49)
2. Eurojackpot-zahlen.eu (für Eurojackpot)
Prüft:
- Vollständigkeit (fehlende Ziehungen)
- Korrektheit (falsche Zahlen)
- Duplikate
"""
import pandas as pd
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
import sys
import os
class DrawVerifier:
"""Verifiziert Ziehungen gegen offizielle Quellen."""
def __init__(self, csv_file: str, lottery_type: str):
self.csv_file = csv_file
self.lottery_type = lottery_type
self.df_local = None
self.df_official = None
self.errors = []
self.warnings = []
self.info = []
def load_local_data(self) -> bool:
"""Lädt lokale CSV-Datei."""
try:
if not os.path.exists(self.csv_file):
self.errors.append(f"❌ Datei existiert nicht: {self.csv_file}")
return False
self.df_local = pd.read_csv(self.csv_file, sep=';')
self.df_local['datum'] = pd.to_datetime(
self.df_local['datum'],
format='%Y-%m-%d',
errors='coerce'
)
self.info.append(f"✅ Lokale Daten: {len(self.df_local)} Ziehungen")
return True
except Exception as e:
self.errors.append(f"❌ Fehler beim Laden: {e}")
return False
def fetch_official_lotto_data(self) -> bool:
"""Holt offizielle Lotto 6aus49 Daten von GitHub Archive."""
try:
url = 'https://johannesfriedrich.github.io/LottoNumberArchive/Lottonumbers_tidy_complete.json'
print("🌐 Lade offizielle Lotto-Daten von GitHub Archive...")
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
}
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
# Parse gruppierte Daten
draws_by_id = {}
for entry in data:
draw_id = entry.get('id')
if draw_id not in draws_by_id:
draws_by_id[draw_id] = {
'date': entry.get('date'),
'numbers': [],
'superzahl': None
}
variable = entry.get('variable')
value = entry.get('value')
if variable == 'Lottozahl':
draws_by_id[draw_id]['numbers'].append(value)
elif variable == 'Superzahl':
draws_by_id[draw_id]['superzahl'] = value
# Konvertiere zu DataFrame
official_draws = []
for draw_id, draw_data in draws_by_id.items():
try:
date_str = draw_data['date']
day, month, year = date_str.split('.')
date_obj = datetime(int(year), int(month), int(day))
numbers = sorted(draw_data['numbers'])
if len(numbers) == 6:
official_draws.append({
'datum': date_obj,
'Z1': numbers[0],
'Z2': numbers[1],
'Z3': numbers[2],
'Z4': numbers[3],
'Z5': numbers[4],
'Z6': numbers[5],
'SZ': draw_data['superzahl']
})
except Exception:
continue
self.df_official = pd.DataFrame(official_draws)
self.info.append(f"✅ Offizielle Daten: {len(self.df_official)} Ziehungen")
return True
except Exception as e:
self.errors.append(f"❌ Fehler beim Laden offizieller Daten: {e}")
return False
def fetch_official_eurojackpot_data(self) -> bool:
"""
Holt offizielle Eurojackpot Daten.
Note: Da es keine vollständige öffentliche API gibt,
beschränken wir uns auf Plausibilitätsprüfungen.
"""
self.warnings.append("⚠️ Keine vollständige offizielle Eurojackpot-API verfügbar")
self.warnings.append(" Nur Plausibilitätsprüfungen möglich")
# Erstelle minimale "offizielle" Daten basierend auf erwarteten Ziehungsterminen
# Dies ist nur für Vollständigkeitsprüfung
if len(self.df_local) > 0:
start_date = self.df_local['datum'].min()
end_date = datetime.now()
expected_dates = []
current = start_date
while current <= end_date:
# Eurojackpot: Dienstag (1) und Freitag (4)
if current.weekday() in [1, 4]:
expected_dates.append(current)
current += timedelta(days=1)
self.df_official = pd.DataFrame({'datum': expected_dates})
self.info.append(f"️ Erwartete Ziehungstermine: {len(expected_dates)}")
return True
def compare_completeness(self) -> List[datetime]:
"""Prüft auf fehlende Ziehungen."""
if self.df_official is None or self.df_local is None:
return []
official_dates = set(self.df_official['datum'].dt.date)
local_dates = set(self.df_local['datum'].dt.date)
missing = official_dates - local_dates
extra = local_dates - official_dates
if missing:
self.errors.append(f"{len(missing)} fehlende Ziehung(en):")
for date in sorted(missing)[:10]:
self.errors.append(f" {date.strftime('%Y-%m-%d')}")
if len(missing) > 10:
self.errors.append(f" ... und {len(missing) - 10} weitere")
if extra:
self.warnings.append(f"⚠️ {len(extra)} zusätzliche Ziehung(en) (nicht in offiziellen Daten):")
for date in sorted(extra)[:5]:
self.warnings.append(f" {date.strftime('%Y-%m-%d')}")
if len(extra) > 5:
self.warnings.append(f" ... und {len(extra) - 5} weitere")
return list(missing)
def compare_accuracy(self) -> int:
"""Vergleicht Zahlenwerte mit offiziellen Daten."""
if self.df_official is None or self.df_local is None:
return 0
if self.lottery_type == 'eurojackpot':
# Keine detaillierten offiziellen Daten verfügbar
return 0
mismatches = 0
# Merge auf Datum
merged = pd.merge(
self.df_local,
self.df_official,
on='datum',
suffixes=('_local', '_official'),
how='inner'
)
if self.lottery_type == 'lotto':
cols_to_check = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
else: # eurojackpot
cols_to_check = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'SZ1', 'SZ2']
for idx, row in merged.iterrows():
mismatch_cols = []
for col in cols_to_check:
local_col = f"{col}_local"
official_col = f"{col}_official"
if local_col in row and official_col in row:
# Vergleiche nur wenn beide Werte vorhanden
if pd.notna(row[local_col]) and pd.notna(row[official_col]):
if row[local_col] != row[official_col]:
mismatch_cols.append(
f"{col}: {int(row[local_col])}{int(row[official_col])}"
)
if mismatch_cols:
mismatches += 1
if mismatches == 1:
self.errors.append("❌ Zahlen-Abweichungen gefunden:")
if mismatches <= 10:
date_str = row['datum'].strftime('%Y-%m-%d')
self.errors.append(f" {date_str}: {', '.join(mismatch_cols)}")
if mismatches > 10:
self.errors.append(f" ... und {mismatches - 10} weitere Abweichungen")
return mismatches
def check_recent_draws(self, days: int = 30) -> None:
"""Prüft besonders die letzten N Tage."""
if self.df_local is None:
return
cutoff = datetime.now() - timedelta(days=days)
recent = self.df_local[self.df_local['datum'] >= cutoff]
self.info.append(f"️ Letzte {days} Tage: {len(recent)} Ziehungen")
if len(recent) == 0:
self.warnings.append(f"⚠️ Keine Ziehungen in den letzten {days} Tagen!")
# Erwartete Anzahl berechnen
if self.lottery_type == 'lotto':
# 2x pro Woche
expected = (days / 7) * 2
else: # eurojackpot
# 2x pro Woche
expected = (days / 7) * 2
if len(recent) < expected * 0.8: # Toleranz 20%
self.warnings.append(
f"⚠️ Weniger Ziehungen als erwartet: {len(recent)} vs. ~{int(expected)}"
)
def verify_all(self) -> bool:
"""Führt komplette Verifikation durch."""
print(f"\n{'='*70}")
print(f" ZIEHUNGS-VERIFIZIERER")
print(f" Typ: {self.lottery_type.upper()}")
print(f"{'='*70}")
print(f"\n📁 Datei: {os.path.basename(self.csv_file)}")
# Lade lokale Daten
if not self.load_local_data():
return False
# Lade offizielle Daten
print()
if self.lottery_type == 'lotto':
if not self.fetch_official_lotto_data():
return False
else: # eurojackpot
if not self.fetch_official_eurojackpot_data():
return False
print(f"\n🔍 VERIFIKATION")
print("="*70)
# Prüfungen
print("Prüfe Vollständigkeit...")
missing = self.compare_completeness()
if self.lottery_type == 'lotto':
print("Prüfe Zahlenwerte...")
mismatches = self.compare_accuracy()
print("Prüfe aktuelle Ziehungen...")
self.check_recent_draws(30)
# Statistiken
print(f"\n📊 STATISTIKEN")
print("="*70)
for info in self.info:
print(info)
# Zusammenfassung
if self.df_local is not None and self.df_official is not None:
if self.lottery_type == 'lotto':
overlap = len(pd.merge(
self.df_local,
self.df_official,
on='datum',
how='inner'
))
if overlap > 0:
print(f"\n{overlap} Ziehungen in beiden Quellen")
# Genauigkeit
if self.lottery_type == 'lotto':
accuracy = ((overlap - (mismatches if 'mismatches' in locals() else 0)) / overlap * 100)
print(f"✅ Genauigkeit: {accuracy:.1f}%")
# Fehler
if self.errors:
print(f"\n❌ FEHLER ({len(self.errors)})")
print("="*70)
for error in self.errors:
print(error)
# Warnungen
if self.warnings:
print(f"\n⚠️ WARNUNGEN ({len(self.warnings)})")
print("="*70)
for warning in self.warnings:
print(warning)
# Ergebnis
print(f"\n{'='*70}")
if not self.errors:
if self.warnings:
print("✅ VERIFIKATION OK - Nur Warnungen")
else:
print("✅ VERIFIKATION ERFOLGREICH - Alle Ziehungen korrekt!")
else:
print("❌ VERIFIKATION FEHLGESCHLAGEN - Fehler gefunden")
print("="*70)
return len(self.errors) == 0
def main():
"""Hauptfunktion."""
# Standard-Dateien
files = {
'lotto': {
'path': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/AlleLottozahlen.csv',
'type': 'lotto'
},
'eurojackpot': {
'path': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv',
'type': 'eurojackpot'
}
}
if len(sys.argv) > 1:
# Einzelne Datei
csv_file = sys.argv[1]
lottery_type = sys.argv[2] if len(sys.argv) > 2 else 'lotto'
verifier = DrawVerifier(csv_file, lottery_type)
success = verifier.verify_all()
sys.exit(0 if success else 1)
else:
# Beide Dateien
print("\n🎲 Verifiziere beide Lottery-Dateien...\n")
results = {}
for name, config in files.items():
if os.path.exists(config['path']):
verifier = DrawVerifier(config['path'], config['type'])
results[name] = verifier.verify_all()
else:
print(f"\n⚠️ {name.upper()}: Datei nicht gefunden")
results[name] = False
# Gesamtergebnis
print("\n" + "="*70)
print(" GESAMTERGEBNIS")
print("="*70)
for name, success in results.items():
status = "" if success else ""
print(f"{status} {name.upper()}")
print("="*70)
all_success = all(results.values())
sys.exit(0 if all_success else 1)
if __name__ == "__main__":
main()
+210
View File
@@ -0,0 +1,210 @@
#!/usr/bin/env python3
"""
Eurojackpot Zahlen-Umschlüsselung
Fügt neue Spalten z1, z2, z3, z4, z5 hinzu mit umschlüsselten Werten:
1-5 → 1, 6-10 → 2, 11-15 → 3, 16-20 → 4, 21-25 → 5,
26-30 → 6, 31-35 → 7, 36-40 → 8, 41-45 → 9, 46-50 → 10
"""
import pandas as pd
def convert_number_to_group(number):
"""Konvertiert eine Zahl (1-50) in eine Gruppe (1-10)."""
if 1 <= number <= 5:
return 1
elif 6 <= number <= 10:
return 2
elif 11 <= number <= 15:
return 3
elif 16 <= number <= 20:
return 4
elif 21 <= number <= 25:
return 5
elif 26 <= number <= 30:
return 6
elif 31 <= number <= 35:
return 7
elif 36 <= number <= 40:
return 8
elif 41 <= number <= 45:
return 9
elif 46 <= number <= 50:
return 10
else:
return 0 # Fehlerfall
def process_number_conversion():
"""Führt die Zahlenumschlüsselung durch."""
# Eingabedatei laden
input_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/data/AlleEurojackpotzahlen.csv"
print("🔢 EUROJACKPOT ZAHLEN-UMSCHLÜSSELUNG")
print("="*50)
# Daten laden
print(f"📁 Lade Daten aus: AlleEurojackpotzahlen.csv")
df = pd.read_csv(input_file, sep=';')
print(f"{len(df)} Ziehungen geladen")
# Umschlüsselungsschema anzeigen
print(f"\n📋 UMSCHLÜSSELUNGSSCHEMA:")
print("="*30)
ranges = [
(1, 5, 1), (6, 10, 2), (11, 15, 3), (16, 20, 4), (21, 25, 5),
(26, 30, 6), (31, 35, 7), (36, 40, 8), (41, 45, 9), (46, 50, 10)
]
for start, end, group in ranges:
print(f"Zahlen {start:2}-{end:2} → Gruppe {group:2}")
# Neue Spalten erstellen
print(f"\n🔄 Erstelle neue Spalten z1, z2, z3, z4, z5...")
source_columns = ['Z1', 'Z2', 'Z3', 'Z4', 'Z5']
target_columns = ['z1', 'z2', 'z3', 'z4', 'z5']
for source_col, target_col in zip(source_columns, target_columns):
df[target_col] = df[source_col].apply(convert_number_to_group)
print(f" {source_col}{target_col}")
# Erste 5 Beispiele anzeigen
print(f"\n📊 BEISPIEL-UMSCHLÜSSELUNGEN (erste 5 Ziehungen):")
print("="*60)
print(f"{'Datum':<12} {'Z1→z1':<8} {'Z2→z2':<8} {'Z3→z3':<8} {'Z4→z4':<8} {'Z5→z5':<8}")
print("-" * 60)
for i in range(min(5, len(df))):
row = df.iloc[i]
datum = row['datum']
conversions = []
for source_col, target_col in zip(source_columns, target_columns):
original = row[source_col]
converted = row[target_col]
conversions.append(f"{original:2}{converted}")
print(f"{datum:<12} {conversions[0]:<8} {conversions[1]:<8} {conversions[2]:<8} {conversions[3]:<8} {conversions[4]:<8}")
# Statistiken der Umschlüsselung
print(f"\n📈 STATISTIKEN DER UMSCHLÜSSELTEN WERTE:")
print("="*45)
# Häufigkeit der Gruppen über alle Positionen
all_converted_values = []
for target_col in target_columns:
all_converted_values.extend(df[target_col].tolist())
from collections import Counter
group_counts = Counter(all_converted_values)
print(f"Verteilung der Gruppen (1-10) über alle Positionen:")
total_values = len(all_converted_values)
for group in range(1, 11):
count = group_counts.get(group, 0)
percentage = (count / total_values) * 100
original_range = f"{(group-1)*5 + 1}-{group*5}"
print(f"Gruppe {group:2} ({original_range:5}): {count:4}x ({percentage:5.1f}%)")
# Statistiken pro Position
print(f"\n📊 VERTEILUNG PRO POSITION:")
print("="*35)
for i, target_col in enumerate(target_columns, 1):
position_counts = Counter(df[target_col])
print(f"\nPosition z{i} ({target_col}):")
for group in range(1, 11):
count = position_counts.get(group, 0)
percentage = (count / len(df)) * 100
print(f" Gruppe {group:2}: {count:3}x ({percentage:4.1f}%)")
# Häufigste Kombinationen der umschlüsselten Werte
print(f"\n🎯 HÄUFIGSTE KOMBINATIONEN (umschlüsselt):")
print("="*45)
# Kombinationen als Strings erstellen
df['kombination_umschluesselt'] = df.apply(
lambda row: f"{row['z1']}-{row['z2']}-{row['z3']}-{row['z4']}-{row['z5']}", axis=1
)
combination_counts = Counter(df['kombination_umschluesselt'])
print(f"Top 15 Kombinationen (z1-z2-z3-z4-z5):")
for i, (combination, count) in enumerate(combination_counts.most_common(15), 1):
percentage = (count / len(df)) * 100
print(f"{i:2}. {combination:15} {count:3}x ({percentage:4.1f}%)")
# Muster-Analyse
print(f"\n🔍 MUSTER-ANALYSE:")
print("="*25)
# Aufsteigende Kombinationen
ascending_count = 0
descending_count = 0
for _, row in df.iterrows():
values = [row[col] for col in target_columns]
if values == sorted(values):
ascending_count += 1
elif values == sorted(values, reverse=True):
descending_count += 1
print(f"Aufsteigende Kombinationen: {ascending_count} ({(ascending_count/len(df)*100):.1f}%)")
print(f"Absteigende Kombinationen: {descending_count} ({(descending_count/len(df)*100):.1f}%)")
# Gleiche Werte
same_values_stats = {}
for num_same in range(2, 6):
count = 0
for _, row in df.iterrows():
values = [row[col] for col in target_columns]
value_counts = Counter(values)
if max(value_counts.values()) >= num_same:
count += 1
same_values_stats[num_same] = count
print(f"Mindestens {num_same} gleiche Werte: {count} ({(count/len(df)*100):.1f}%)")
# Bereiche der umschlüsselten Werte
print(f"\n📋 BEREICHSANALYSE (umschlüsselt):")
print("="*35)
# Niedrig (1-3), Mittel (4-7), Hoch (8-10)
for i, target_col in enumerate(target_columns, 1):
low_count = sum(1 for val in df[target_col] if 1 <= val <= 3)
mid_count = sum(1 for val in df[target_col] if 4 <= val <= 7)
high_count = sum(1 for val in df[target_col] if 8 <= val <= 10)
low_pct = (low_count / len(df)) * 100
mid_pct = (mid_count / len(df)) * 100
high_pct = (high_count / len(df)) * 100
print(f"z{i}: Niedrig(1-3)={low_pct:4.1f}% | Mittel(4-7)={mid_pct:4.1f}% | Hoch(8-10)={high_pct:4.1f}%")
# Ausgabedatei speichern
output_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Eurojackpot/AlleEurojackpotzahlen_umschluesselt.csv"
print(f"\n💾 DATEI SPEICHERN:")
print("="*25)
# Spalten neu ordnen (Original + neue Spalten)
column_order = ['tag', 'datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'z1', 'z2', 'z3', 'z4', 'z5', 'SZ1', 'SZ2', 'kombination_umschluesselt']
# Prüfen welche Spalten existieren
available_columns = [col for col in column_order if col in df.columns]
df_output = df[available_columns]
df_output.to_csv(output_file, sep=';', index=False)
print(f"✅ Umschlüsselte Daten gespeichert: AlleEurojackpotzahlen_umschluesselt.csv")
print(f"📊 Anzahl Spalten: {len(df_output.columns)}")
print(f"📈 Anzahl Zeilen: {len(df_output)}")
print(f"\n🔍 NEUE SPALTEN:")
for col in ['z1', 'z2', 'z3', 'z4', 'z5', 'kombination_umschluesselt']:
if col in df_output.columns:
print(f"{col}")
return df_output
if __name__ == "__main__":
result_df = process_number_conversion()