cbazzaandClaude Sonnet 4.5
f6106b8333
This project includes multiple AI/ML-based lottery number generators for German Lotto 6aus49, including pattern analysis, weighted predictions, and hybrid approaches. Features automated weekly tip generation, performance tracking, and Telegram bot integration. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
500 lines
16 KiB
Python
500 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Lotto 6aus49 Data Updater für lotto.de
|
|
|
|
Web-Scraper für https://www.lotto.de/lotto-6aus49/lottozahlen
|
|
|
|
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
|
|
import time
|
|
|
|
|
|
class LottoWebUpdater:
|
|
"""Updater für lotto.de Web-Scraping"""
|
|
|
|
def __init__(self, data_file: str):
|
|
self.data_file = data_file
|
|
self.url = "https://www.lotto.de/lotto-6aus49/lottozahlen"
|
|
self.backup_file = None
|
|
self.df_existing = None
|
|
|
|
print("🔄 LOTTO 6AUS49 WEB 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', 'Z6', 'SZ']
|
|
)
|
|
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")
|
|
parent_dir = os.path.dirname(self.data_file)
|
|
lotto_dir = os.path.join(parent_dir, "Lotto")
|
|
backup_dir = os.path.join(lotto_dir, "data", "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, max_pages: int = 5) -> List[Dict]:
|
|
"""
|
|
Scrapt Ziehungen von lotto.de
|
|
|
|
Args:
|
|
max_pages: Maximale Anzahl Seiten (jede Seite ~20 Ziehungen)
|
|
|
|
Returns:
|
|
Liste von Ziehungen
|
|
"""
|
|
print(f"\n🌐 Lade Daten von {self.url}...")
|
|
print(f" 📄 Lade bis zu {max_pages} Seiten...")
|
|
|
|
all_draws = []
|
|
|
|
try:
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
|
|
}
|
|
|
|
for page in range(1, max_pages + 1):
|
|
url = f"{self.url}?page={page}" if page > 1 else self.url
|
|
|
|
print(f"\n 📄 Seite {page}/{max_pages}...")
|
|
|
|
response = requests.get(url, headers=headers, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
soup = BeautifulSoup(response.content, 'html.parser')
|
|
|
|
# Finde Ziehungen auf dieser Seite
|
|
page_draws = self._extract_draws_from_page(soup, page)
|
|
|
|
if not page_draws:
|
|
print(f" ⚠️ Keine Ziehungen gefunden - Abbruch")
|
|
break
|
|
|
|
all_draws.extend(page_draws)
|
|
print(f" ✅ {len(page_draws)} Ziehungen extrahiert")
|
|
|
|
# Rate limiting
|
|
if page < max_pages:
|
|
time.sleep(1)
|
|
|
|
print(f"\n ✅ Gesamt: {len(all_draws)} Ziehungen von {page} Seite(n)")
|
|
return all_draws
|
|
|
|
except requests.RequestException as e:
|
|
print(f" ❌ Netzwerkfehler: {e}")
|
|
return all_draws # Return was bisher geladen wurde
|
|
except Exception as e:
|
|
print(f" ❌ Fehler: {e}")
|
|
return all_draws
|
|
|
|
def _extract_draws_from_page(self, soup: BeautifulSoup, page_num: int) -> List[Dict]:
|
|
"""
|
|
Extrahiert Ziehungen von einer Seite.
|
|
|
|
Args:
|
|
soup: BeautifulSoup Objekt
|
|
page_num: Seitennummer
|
|
|
|
Returns:
|
|
Liste von Ziehungen
|
|
"""
|
|
draws = []
|
|
|
|
try:
|
|
# Verschiedene Selektoren versuchen
|
|
# 1. Versuche mit data-attribute
|
|
result_divs = soup.find_all('div', class_=lambda x: x and 'result' in x.lower())
|
|
|
|
if not result_divs:
|
|
# 2. Versuche table
|
|
result_divs = soup.find_all('tr', class_=lambda x: x and ('draw' in x.lower() or 'result' in x.lower()))
|
|
|
|
if not result_divs:
|
|
# 3. Versuche generische Container
|
|
result_divs = soup.find_all('div', class_=lambda x: x and 'drawing' in x.lower())
|
|
|
|
print(f" 🔍 {len(result_divs)} potentielle Ergebnis-Container gefunden")
|
|
|
|
for i, elem in enumerate(result_divs):
|
|
try:
|
|
# Extrahiere Datum
|
|
date_elem = elem.find('time') or elem.find(class_=lambda x: x and 'date' in x.lower())
|
|
if not date_elem:
|
|
# Versuche direkten Text
|
|
date_text = self._extract_date_text(elem.get_text())
|
|
else:
|
|
date_text = date_elem.get_text().strip()
|
|
|
|
date_obj = self._parse_date(date_text)
|
|
|
|
# Extrahiere Zahlen - verschiedene Ansätze
|
|
numbers = self._extract_numbers(elem)
|
|
|
|
if len(numbers) < 7: # Mindestens 6 Hauptzahlen + 1 Superzahl
|
|
continue
|
|
|
|
main_numbers = sorted(numbers[:6])
|
|
superzahl = numbers[6]
|
|
|
|
draw = {
|
|
'datum': date_obj,
|
|
'Z1': main_numbers[0],
|
|
'Z2': main_numbers[1],
|
|
'Z3': main_numbers[2],
|
|
'Z4': main_numbers[3],
|
|
'Z5': main_numbers[4],
|
|
'Z6': main_numbers[5],
|
|
'SZ': superzahl
|
|
}
|
|
|
|
if self._validate_draw(draw):
|
|
draws.append(draw)
|
|
else:
|
|
print(f" ⚠️ Element {i+1}: Validierung fehlgeschlagen")
|
|
|
|
except Exception as e:
|
|
continue
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ Fehler beim Extrahieren: {e}")
|
|
|
|
return draws
|
|
|
|
def _extract_date_text(self, text: str) -> str:
|
|
"""Extrahiert Datum aus Text."""
|
|
import re
|
|
|
|
# Suche nach Datum im Format DD.MM.YYYY oder YYYY-MM-DD
|
|
patterns = [
|
|
r'(\d{2}\.\d{2}\.\d{4})',
|
|
r'(\d{4}-\d{2}-\d{2})',
|
|
r'(\d{1,2}\.\s*\w+\s*\d{4})' # z.B. "22. Januar 2025"
|
|
]
|
|
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text)
|
|
if match:
|
|
return match.group(1)
|
|
|
|
return text.strip()
|
|
|
|
def _extract_numbers(self, elem) -> List[int]:
|
|
"""Extrahiert Zahlen aus Element."""
|
|
numbers = []
|
|
|
|
# 1. Versuche mit span/div mit Klassen wie 'ball', 'number', etc.
|
|
number_elems = elem.find_all(class_=lambda x: x and any(c in x.lower() for c in ['ball', 'number', 'zahl']))
|
|
|
|
if number_elems:
|
|
for num_elem in number_elems:
|
|
try:
|
|
text = num_elem.get_text().strip()
|
|
# Extrahiere nur Zahlen
|
|
import re
|
|
nums = re.findall(r'\d+', text)
|
|
if nums:
|
|
numbers.append(int(nums[0]))
|
|
except:
|
|
continue
|
|
|
|
# 2. Fallback: Alle Zahlen aus Text extrahieren
|
|
if len(numbers) < 7:
|
|
import re
|
|
text = elem.get_text()
|
|
# Finde alle Zahlen
|
|
all_nums = re.findall(r'\b(\d+)\b', text)
|
|
|
|
# Filtere plausible Lotto-Zahlen (1-49 für Hauptzahlen, 0-9 für Superzahl)
|
|
plausible = []
|
|
for n in all_nums:
|
|
num = int(n)
|
|
if 1 <= num <= 49:
|
|
plausible.append(num)
|
|
elif 0 <= num <= 9 and len(plausible) >= 6:
|
|
# Könnte Superzahl sein
|
|
plausible.append(num)
|
|
|
|
if len(plausible) >= 7:
|
|
numbers = plausible[:7]
|
|
|
|
return numbers
|
|
|
|
def _parse_date(self, date_str: str) -> datetime:
|
|
"""Parst Datum."""
|
|
import re
|
|
|
|
# Clean up
|
|
date_str = date_str.strip()
|
|
|
|
# Format 1: DD.MM.YYYY
|
|
try:
|
|
return datetime.strptime(date_str, '%d.%m.%Y')
|
|
except ValueError:
|
|
pass
|
|
|
|
# Format 2: YYYY-MM-DD
|
|
try:
|
|
return datetime.strptime(date_str, '%Y-%m-%d')
|
|
except ValueError:
|
|
pass
|
|
|
|
# Format 3: DD. MONAT YYYY (z.B. "22. Januar 2025")
|
|
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
|
|
}
|
|
|
|
match = re.search(r'(\d{1,2})\.\s*(\w+)\s*(\d{4})', date_str.lower())
|
|
if match:
|
|
day = int(match.group(1))
|
|
month_str = match.group(2)
|
|
year = int(match.group(3))
|
|
|
|
if month_str in months_de:
|
|
return datetime(year, months_de[month_str], day)
|
|
|
|
# Fallback: Aktuelles Datum
|
|
print(f" ⚠️ Konnte Datum nicht parsen: {date_str}")
|
|
return datetime.now()
|
|
|
|
def _validate_draw(self, draw: Dict) -> bool:
|
|
"""Validiert eine Ziehung."""
|
|
try:
|
|
required_fields = ['datum', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'SZ']
|
|
if not all(field in draw for field in required_fields):
|
|
return False
|
|
|
|
# Hauptzahlen (1-49)
|
|
main_numbers = [draw[f'Z{i}'] for i in range(1, 7)]
|
|
if not all(isinstance(n, int) and 1 <= n <= 49 for n in main_numbers):
|
|
return False
|
|
|
|
if len(set(main_numbers)) != 6:
|
|
return False
|
|
|
|
# Superzahl (0-9)
|
|
if not isinstance(draw['SZ'], int) or not 0 <= draw['SZ'] <= 9:
|
|
return False
|
|
|
|
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."""
|
|
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', 'Z6', 'SZ']
|
|
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, max_pages: int = 5, create_backup: bool = True) -> bool:
|
|
"""
|
|
Führt komplettes Update durch.
|
|
|
|
Args:
|
|
max_pages: Maximale Anzahl Seiten zum Scrapen
|
|
create_backup: Backup vor Update 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()
|
|
|
|
new_draws = self.fetch_draws_from_web(max_pages=max_pages)
|
|
|
|
if not new_draws:
|
|
print("\n❌ Keine Daten von Website geladen")
|
|
print("💡 Tipp: Verwende simple_update.py für manuelle Eingabe")
|
|
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(" LOTTO 6AUS49 WEB UPDATER")
|
|
print(" Quelle: lotto.de")
|
|
print("=" * 70)
|
|
print()
|
|
|
|
default_file = "/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/Lotto/data/AlleLottozahlen.csv"
|
|
|
|
if len(sys.argv) > 1:
|
|
data_file = sys.argv[1]
|
|
max_pages = int(sys.argv[2]) if len(sys.argv) > 2 else 5
|
|
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()
|
|
|
|
pages_input = input("\nAnzahl Seiten laden (1-10, default=5): ").strip()
|
|
max_pages = int(pages_input) if pages_input else 5
|
|
|
|
print()
|
|
|
|
updater = LottoWebUpdater(data_file)
|
|
success = updater.update(max_pages=max_pages)
|
|
|
|
sys.exit(0 if success else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|