Spaces:
Sleeping
Sleeping
File size: 3,845 Bytes
2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 2d25973 62745f2 | 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 | import os
from supabase import create_client, Client
import bcrypt
from dotenv import load_dotenv
# Force environment reload safely
load_dotenv()
SUPABASE_URL = os.environ.get("SUPABASE_URL")
SUPABASE_KEY = os.environ.get("SUPABASE_KEY")
if not SUPABASE_URL or not SUPABASE_KEY:
raise ValueError("CRITICAL: Missing SUPABASE_URL or SUPABASE_KEY inside the .env file! Phase 7 requires this constraint.")
def get_supabase() -> Client:
return create_client(SUPABASE_URL, SUPABASE_KEY)
def create_user(email, password=None, provider="email"):
user = get_user_by_email(email)
if user:
return False, "Email already registered."
password_hash = None
if password:
password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
try:
res = get_supabase().table("users").insert({
"email": email.lower(),
"password_hash": password_hash,
"provider": provider
}).execute()
if provider != "email":
return res.data[0]["id"]
return True, "Account successfully created! Please log in."
except Exception as e:
if provider != "email": return None
return False, f"Database error: {e}"
def authenticate_user(email, password):
user = get_user_by_email(email)
if not user:
return False, "User not found."
stored_hash = user.get("password_hash")
if stored_hash:
if bcrypt.checkpw(password.encode('utf-8'), stored_hash.encode('utf-8')):
return True, user["id"]
else:
return False, "Incorrect password."
else:
return False, f"Account uses OAuth. Please login with {user.get('provider')}."
def get_user_by_email(email):
res = get_supabase().table("users").select("*").eq("email", email.lower()).execute()
if res.data and len(res.data) > 0:
return res.data[0]
return None
def get_user_by_id(user_id):
res = get_supabase().table("users").select("*").eq("id", user_id).execute()
if res.data and len(res.data) > 0:
return res.data[0]
return None
def authenticate_oauth_user(email, provider):
user = get_user_by_email(email)
if user:
return True, user["id"]
# create_user returns (True, Msg) or (False, Msg) or raw ID if not email.
# Actually create_user returns ID directly if provider != "email":
# Let's cleanly orchestrate the ID return:
uid = create_user(email, provider=provider)
if uid:
return True, uid
return False, "Failed to authenticate with provider."
def create_session(user_id, filename, pdf_text, notes):
payload = {
"user_id": user_id,
"filename": filename,
"pdf_text": pdf_text,
"notes": notes,
"chat_history": []
}
res = get_supabase().table("sessions").insert(payload).execute()
return res.data[0]["id"]
def get_all_sessions(user_id):
# Retrieve all columns inherently required by `app.py` list map loops
res = get_supabase().table("sessions").select("*").eq("user_id", user_id).order("timestamp", desc=True).execute()
return res.data if res.data else []
def get_session(session_id):
res = get_supabase().table("sessions").select("*").eq("id", session_id).execute()
if res.data and len(res.data) > 0:
return res.data[0]
return None
def save_chat_message(session_id, role, content):
session_data = get_session(session_id)
if session_data:
chat_history = session_data.get("chat_history", [])
chat_history.append({"role": role, "content": content})
get_supabase().table("sessions").update({"chat_history": chat_history}).eq("id", session_id).execute()
def delete_session(session_id):
get_supabase().table("sessions").delete().eq("id", session_id).execute()
|