Spaces:
Runtime error
Runtime error
| 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""" | |
| <html> | |
| <body style="font-family: 'Segoe UI', Arial, sans-serif; background-color: #0a0a12; color: #ffffff; padding: 40px; margin: 0;"> | |
| <div style="max-width: 500px; margin: auto; background-color: #0a0a12; border: 1px solid #16213e; padding: 20px; border-radius: 10px;"> | |
| <h1 style="color: #e94560; font-size: 28px; margin-bottom: 25px; letter-spacing: 1px;">Pilot Verification</h1> | |
| <p style="font-size: 16px; color: #cccccc; margin-bottom: 30px;">Your authentication code for the Static Defender Hangar is:</p> | |
| <div style="background-color: #16213e; | |
| border: 2px solid #e94560; | |
| padding: 15px 25px; | |
| display: inline-block; | |
| border-radius: 4px; | |
| margin-bottom: 30px;"> | |
| <span style="color: #ffffff; | |
| font-size: 36px; | |
| font-weight: 900; | |
| letter-spacing: 8px; | |
| font-family: monospace;">{otp_code}</span> | |
| </div> | |
| <p style="font-size: 14px; color: #888888; border-top: 1px solid #16213e; padding-top: 20px;"> | |
| Enter this code to initialize your pilot profile. Mission Control is standing by. | |
| </p> | |
| </div> | |
| </body> | |
| </html> | |
| """ | |
| # 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 [] | |