FDJ-SLAYER/main.py

"""
This script generates random draws for EuroMillions using various entropy sources.
"""

import os
import time
import hashlib
import secrets
import random
import socket
import platform
import psutil
import uuid
import threading
import datetime
import multiprocessing
from progress.bar import Bar
import openmeteo_requests
import requests_cache
import numpy as _
from retry_requests import retry
from constants import *

def _fetch_weather_api(params):
    """Internal function to make Open-Meteo API requests"""
    cache_session = requests_cache.CachedSession('.cache', expire_after=60)
    retry_session = retry(cache_session, retries=2, backoff_factor=0.2)
    openmeteo = openmeteo_requests.Client(session=retry_session)
    return openmeteo.weather_api(OPENMETEO_API_URL, params=params)

def _get_weather_data():
    """Internal function that retrieves weather data from API"""
    params = {
        "latitude": random.uniform(-70, 70),
        "longitude": random.uniform(-180, 180),
        "hourly": random.sample(OPENMETEO_HOURLY_PARAMS, random.randint(3, 6)),
        "timezone": "auto"
    }

    responses = _fetch_weather_api(params)
    entropy_values = []
    for i in range(len(params["hourly"])):
        var_values = responses[0].Hourly().Variables(i).ValuesAsNumpy()
        entropy_values.extend(var_values.tolist())

    weather_str = "".join([str(v) for v in entropy_values])
    return hashlib.sha256(weather_str.encode()).hexdigest()

def get_weather_entropy():
    """Retrieves weather data as additional source of entropy"""
    try:
        return _get_weather_data()
    except Exception as e:
        print(f"Error retrieving weather data: {e}")
        fallback = f"weather_fallback_{time.time()}_{random.getrandbits(64)}"
        return hashlib.sha256(fallback.encode()).hexdigest()

def get_static_entropy_pool():
    """Creates a base entropy pool with sources that don't change rapidly"""
    weather_entropy = get_weather_entropy()
    return [
        str(os.urandom(32)),
        str(secrets.token_bytes(32)),
        socket.gethostname(),
        str(platform.system_alias(platform.system(), platform.release(), platform.version())),
        str(uuid.getnode()),
        str(multiprocessing.cpu_count()),
        "".join([str(v) for v in psutil.disk_partitions()]),
        str(hash(frozenset(os.environ.items()))),
        weather_entropy,
    ]

def get_dynamic_entropy_pool():
    """Retrieves dynamic data to add entropy"""
    return [
        str(time.time()),
        hashlib.sha256(str(time.perf_counter()).encode()).hexdigest(),
        str(os.getpid()),
        str(psutil.cpu_percent(interval=0.01)),
        str(psutil.virtual_memory().percent),
        str(psutil.disk_usage('/').percent),
        str(random.getrandbits(256)),
        str(datetime.datetime.now().microsecond),
        str(threading.active_count()),
        str(sum(psutil.cpu_times())),
        str(psutil.net_io_counters().bytes_sent if hasattr(psutil, 'net_io_counters') else 0),
        str(id({})),
    ]

def generate_seed(base_pool=None):
    """Generates a random seed by combining different entropy sources"""
    entropy_sources = base_pool.copy() if base_pool else []
    entropy_sources.extend(get_dynamic_entropy_pool())
    random.shuffle(entropy_sources)

    entropy_str = "".join(entropy_sources)
    intermediate_hash = hashlib.sha512(entropy_str.encode()).digest()
    intermediate_hash = hashlib.blake2b(intermediate_hash).digest()

    final_hash = hashlib.sha256(intermediate_hash).hexdigest()
    return int(final_hash, 16)

def make_draw(base_pool=None):
    """Generates a draw with N numbers and M stars using a random seed"""
    seed = generate_seed(base_pool)
    random.seed(seed)
    numbers = random.sample(range(1, MAX_NUMBER + 1), NUMBER_OF_NUMBERS)
    stars = random.sample(range(1, MAX_STAR + 1), NUMBER_OF_STARS)
    return {
        "seed": seed,
        "numbers": sorted(numbers),
        "stars": sorted(stars)
    }

def generate_draws(num_draws):
    """Generates multiple draws sequentially"""
    base_pool = get_static_entropy_pool()
    bar = Bar('Progress', max=num_draws, suffix='%(percent)d%%')

    draws = []
    for _ in range(num_draws):
        draws.append(make_draw(base_pool))
        time.sleep(0.01)
        bar.next()

    bar.finish()
    return draws

def display_draw(draw, index=None, title="DRAW"):
    """Displays a draw in a formatted way"""
    print(f"\n===== {title} =====")
    if index is not None:
        print(f"Draw #{index + 1}")
    print(f"Numbers: {draw['numbers']}")
    print(f"Stars: {draw['stars']}")
    print(f"Seed used: {draw['seed']}")

def display_additional_draws(draws, displayed_draws):
    """Displays additional random draws at the user's request."""
    while len(displayed_draws) < len(draws):
        display_another = input("\nDo you want to see another random draw? (y/n): ").lower().strip()
        if display_another not in ['o', 'oui', 'y', 'yes', '']:
            break

        available_indices = [i for i in range(len(draws)) if i not in displayed_draws]
        if not available_indices:
            print("All draws have already been displayed!")
            break

        new_index = random.choice(available_indices)
        new_draw = draws[new_index]
        displayed_draws.add(new_index)

        display_draw(new_draw, new_index, "ANOTHER RANDOM DRAW")

if __name__ == "__main__":
    print(f"Generating {NUMBER_OF_DRAWS} different draws...")
    draws = generate_draws(NUMBER_OF_DRAWS)

    base_pool = get_static_entropy_pool()
    random.seed(generate_seed(base_pool))
    chosen_draw = random.choice(draws)
    displayed_draws = {draws.index(chosen_draw)}

    display_draw(chosen_draw, title="FINAL RESULT")
    print("Draw selected from among the", NUMBER_OF_DRAWS, "generated")
    display_additional_draws(draws, displayed_draws)