Spaces:
Runtime error
Runtime error
File size: 9,152 Bytes
2f9adb5 ac69fce 2f9adb5 ac69fce edbe9ed ac69fce edbe9ed ac69fce edbe9ed ac69fce 2f9adb5 ac69fce 2f9adb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | 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 []
|