Lotto-Tip-Generator/scripts/utils/verify_draws.py

#!/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/Lotto/data/AlleLottozahlen.csv',
            'type': 'lotto'
        },
        'eurojackpot': {
            'path': '/Users/sebastianfrohlich/Library/Mobile Documents/com~apple~CloudDocs/Jupyter Notebooks/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()