import gradio as gr
import json
import os
from datetime import datetime
from dotenv import load_dotenv
from supabase import create_client, Client
from pinecone import Pinecone
from sentence_transformers import SentenceTransformer
from typing import List, Dict
load_dotenv()
SUPABASE_URL = os.getenv("DB_URL")
SUPABASE_KEY = os.getenv("DB_KEY")
pinecone_api_key = os.getenv("PINECONE")
supabase_client = create_client(SUPABASE_URL, SUPABASE_KEY)
pc = Pinecone(api_key=pinecone_api_key)
index = pc.Index("focus-guru")
model = SentenceTransformer("all-MiniLM-L6-v2")
def ingest_user_progress(
supabase_client: Client,
user_id: int,
video_id: str,
rating: float,
time_spent: int,
play_count: int,
completed: bool
):
data = {
"user_id": user_id,
"video_id": video_id,
"rating": rating,
"time_spent": time_spent,
"play_count": play_count,
"completed": completed,
"updated_at": datetime.now().isoformat()
}
response = supabase_client.table("user_progress").insert(data, upsert=True).execute()
return response.data
def gradio_ingest(user_input):
try:
data = json.loads(user_input)
user_id = int(data.get("user_id", 0))
video_id = data.get("video_id", "")
rating = float(data.get("rating", 0))
time_spent = int(data.get("time_spent", 0))
play_count = int(data.get("play_count", 0))
completed = bool(data.get("completed", False))
except Exception as e:
return f"Error parsing input: {e}
"
res = ingest_user_progress(supabase_client, user_id, video_id, rating, time_spent, play_count, completed)
return f"Ingested data: {res}
"
def recommend_playlists_by_package_and_module(assessment_output, index, model):
report_text = assessment_output.get("report", "")
packages = assessment_output.get("package", [])
modules = ["Nutrition", "Exercise", "Meditation"]
recommendations = {}
if not report_text:
for pkg in packages:
recommendations[pkg] = {mod: {"title": "No playlist found", "description": ""} for mod in modules}
return recommendations
query_embedding = model.encode(report_text, convert_to_numpy=True).tolist()
for pkg in packages:
recommendations[pkg] = {}
for mod in modules:
filter_dict = {"type": "playlist", "Package": pkg, "Module": mod}
results = index.query(vector=query_embedding, top_k=1, include_metadata=True, filter=filter_dict)
if results["matches"]:
match = results["matches"][0]
metadata = match["metadata"]
title = metadata.get("Playlist Name", "Unknown Playlist")
description = metadata.get("Description", "")
recommendations[pkg][mod] = {"title": title, "description": description}
else:
recommendations[pkg][mod] = {"title": "No playlist found", "description": ""}
return recommendations
def gradio_recommend_playlist(input_json):
try:
assessment_data = json.loads(input_json)
except json.JSONDecodeError:
return "Error: Invalid JSON format
"
if "package" not in assessment_data or "report" not in assessment_data:
return "Error: Missing 'package' or 'report' field
"
recs = recommend_playlists_by_package_and_module(assessment_data, index, model)
html_output = ""
for pkg, mod_recs in recs.items():
html_output += f"
{pkg} Package
"
for mod, rec in mod_recs.items():
html_output += f"""
{mod} Module
{rec['title']}
{rec['description']}
"""
html_output += "
"
html_output += "
"
# Use black cards with white text and orange border for visibility
for rec in recs:
html_output += f"""
{rec['title']}
{rec['description']}
Score: {rec['score']:.2f}
"""
html_output += "