import psycopg2 from psycopg2 import sql import sqlite3 import hashlib import os import sib_api_v3_sdk from sib_api_v3_sdk.rest import ApiException class AuthManager: def __init__(self): self.db_uri = os.getenv('SUPABASE_URL') # Initialize Brevo Configuration self.configuration = sib_api_v3_sdk.Configuration() self.configuration.api_key['api-key'] = os.getenv('BREVO_API_KEY') self.api_instance = sib_api_v3_sdk.TransactionalEmailsApi(sib_api_v3_sdk.ApiClient(self.configuration)) def send_otp_via_brevo(self, receiver_email, otp_code): """Sends a stylized transaction email via Brevo API.""" subject = "Mission Authorization: Your Pilot OTP" # MISSION-READY STYLING html_content = f"""

Pilot Verification

Your authentication code for the Static Defender Hangar is:

{otp_code}

Enter this code to initialize your pilot profile. Mission Control is standing by.

""" # IMPORTANT: Change this email to your verified Brevo sender email sender = { "name": "Mission Control", "email": "dhruvasthana7@gmail.com" } to = [{ "email": receiver_email }] send_smtp_email = sib_api_v3_sdk.SendSmtpEmail( to=to, html_content=html_content, sender=sender, subject=subject ) try: self.api_instance.send_transac_email(send_smtp_email) return True except ApiException as e: print(f"❌ Brevo Transmission Error: {e}") return False def _init_db(self): with self._get_connection() as conn: with conn.cursor() as cur: # Table 1: Users cur.execute( """ CREATE TABLE IF NOT EXISTS users ( username TEXT PRIMARY KEY, password TEXT, email TEXT, high_score INTEGER DEFAULT 0 ) """ ) # Table 2: Game Sessions cur.execute(""" CREATE TABLE IF NOT EXISTS game_sessions ( id SERIAL PRIMARY KEY, username TEXT REFERENCES users(username) ON UPDATE CASCADE, score INTEGER, duration INTEGER, played_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() def _get_connection(self): return psycopg2.connect(self.db_uri) def _hash_pw(self, password): return hashlib.sha256(password.encode()).hexdigest() def signup(self, username, password, email): try: with self._get_connection() as conn: with conn.cursor() as cur: # Note the use of %s instead of ? for PostgreSQL cur.execute( "INSERT INTO users (username, password, email) VALUES (%s, %s, %s)", (username, self._hash_pw(password), email), ) conn.commit() return True, "Registry complete. Welcome to the Fleet!" except psycopg2.errors.UniqueViolation: return False, "Callsign taken. This galaxy isn't big enough for two of you." except Exception as e: return False, f"Database Error: {str(e)}" def email_exists(self, email): try: with self._get_connection() as conn: with conn.cursor() as cur: cur.execute("SELECT username FROM users WHERE email = %s", (email,)) return cur.fetchone() is not None except Exception: return False def login(self, username, password): try: with self._get_connection() as conn: with conn.cursor() as cur: cur.execute( "SELECT * FROM users WHERE username = %s AND password = %s", (username, self._hash_pw(password)), ) user = cur.fetchone() if user: return True, f"Access Granted. Welcome back, Commander {username}." return False, "Negative. Invalid username/password combination." except Exception as e: return False, "Comms link to database failed." # Added a placeholder for Forgot Password requirement def reset_password(self, email, new_username, new_password): try: hashed_pw = self._hash_pw(new_password) with self._get_connection() as conn: with conn.cursor() as cur: # Update both fields based on the verified email cur.execute( "UPDATE users SET username = %s, password = %s WHERE email = %s", (new_username, hashed_pw, email), ) conn.commit() return True, "Pilot identity and memory bank updated!" except psycopg2.errors.UniqueViolation: return False, "That new username is already taken by another pilot." except Exception as e: return False, f"Update failed: {str(e)}" def get_leaderboard(self, limit=5): """Fetches the top players based on high_score.""" try: with self._get_connection() as conn: with conn.cursor() as cur: cur.execute( "SELECT username, high_score FROM users ORDER BY high_score DESC LIMIT %s", (limit,), ) return cur.fetchall() except Exception as e: print(f"Leaderboard Error: {e}") return [] def save_session(self, username, score, duration): try: with self._get_connection() as conn: with conn.cursor() as cur: # A. Record the session as usual cur.execute( "INSERT INTO game_sessions (username, score, duration) VALUES (%s, %s, %s)", (username, score, duration), ) # B. Update the High Score in the users table ONLY IF the new score is higher cur.execute( """ UPDATE users SET high_score = GREATEST(high_score, %s) WHERE username = %s """, (score, username), ) conn.commit() return True except Exception as e: print(f"❌ SQL ERROR in save_session: {e}") return False def get_session_history(self, username, limit=10): try: with self._get_connection() as conn: with conn.cursor() as cur: cur.execute( """ SELECT score, TO_CHAR(played_at, 'HH24:MI') FROM game_sessions WHERE username = %s ORDER BY played_at DESC LIMIT %s """, (username, limit), ) rows = cur.fetchall() # Return even if only 1 row exists return rows[::-1] except Exception as e: print(f"❌ SQL ERROR in get_session_history: {e}") return []