app.py

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"<p style='color: red;'>Error parsing input: {e}</p>"
    res = ingest_user_progress(supabase_client, user_id, video_id, rating, time_spent, play_count, completed)
    return f"<p style='color: green;'>Ingested data: {res}</p>"

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 "<p style='color: red;'>Error: Invalid JSON format</p>"
    if "package" not in assessment_data or "report" not in assessment_data:
        return "<p style='color: red;'>Error: Missing 'package' or 'report' field</p>"
    recs = recommend_playlists_by_package_and_module(assessment_data, index, model)
    html_output = "<div style='padding: 20px; font-family: Arial, sans-serif;'>"
    for pkg, mod_recs in recs.items():
        html_output += f"<h2>{pkg} Package</h2><div style='display: flex; flex-wrap: wrap; gap: 20px;'>"
        for mod, rec in mod_recs.items():
            html_output += f"""

            <div style="border: 1px solid #ccc; border-radius: 8px; padding: 15px; width: 260px;">

                <h3>{mod} Module</h3>

                <strong>{rec['title']}</strong>

                <p>{rec['description']}</p>

            </div>

            """
        html_output += "</div>"
    html_output += "</div>"
    return html_output

def recommend_videos(user_id: int, K: int = 5, M: int = 10, N: int = 5) -> Dict:
    response = supabase_client.table("user_progress").select("video_id, rating, completed, play_count, videos!inner(playlist_id)").eq("user_id", user_id).execute()
    interactions = response.data
    if not interactions:
        return {
            "note": "No interactions recorded for this user yet. Please watch or rate some videos.",
            "recommendations": []
        }
    for inter in interactions:
        rating = inter["rating"] if inter["rating"] is not None else 0
        completed_val = 1 if inter["completed"] else 0
        play_count = inter["play_count"]
        inter["engagement"] = rating + 2 * completed_val + play_count
    top_videos = sorted(interactions, key=lambda x: x["engagement"], reverse=True)[:K]
    watched_completed_videos = {i["video_id"] for i in interactions if i["completed"]}
    watched_incomplete_videos = {i["video_id"] for i in interactions if not i["completed"]}
    candidates = {}
    for top_video in top_videos:
        query_id = f"video_{top_video['video_id']}"
        response = index.query(id=query_id, top_k=M + 1, include_metadata=True)
        for match in response.get("matches", []):
            if match["id"] == query_id:
                continue
            metadata = match.get("metadata", {})
            vid = metadata.get("vid")
            if not vid:
                continue
            if vid in watched_completed_videos:
                continue
            similarity = match["score"]
            pid = metadata.get("PID")
            boost = 1.1 if pid == top_video["videos"]["playlist_id"] else 1.0
            partial_score = top_video["engagement"] * similarity * boost
            if vid in candidates:
                candidates[vid]["total_score"] += partial_score
            else:
                candidates[vid] = {"total_score": partial_score, "metadata": metadata}
    sorted_candidates = sorted(candidates.items(), key=lambda x: x[1]["total_score"], reverse=True)[:N]
    recommendations = []
    for vid, data in sorted_candidates:
        metadata = data["metadata"]
        video_title = metadata.get("video_title", "Untitled Video")
        if vid in watched_incomplete_videos:
            video_title += " (Incomplete)"
        recommendations.append({
            "video_id": vid,
            "title": video_title,
            "description": metadata.get("video_description", ""),
            "score": data["total_score"]
        })
    note_text = "Based on your engagement, here are some recommended videos from the same playlist."
    return {"note": note_text, "recommendations": recommendations}

def gradio_recommend_videos(user_id_input):
    try:
        user_id = int(user_id_input)
    except Exception as e:
        return f"Error: {e}", ""
    result = recommend_videos(user_id)
    note_text = result["note"]
    recs = result["recommendations"]
    if not recs:
        return note_text, ""
    html_output = "<div>"
    # Use black cards with white text and orange border for visibility
    for rec in recs:
        html_output += f"""

        <div style="background: #000; color: #fff; border: 2px solid orange; border-radius: 8px; margin-bottom: 10px; padding: 15px;">

            <h3 style="margin-top: 0;">{rec['title']}</h3>

            <p style="margin: 0;">{rec['description']}</p>

            <p style="margin: 0;"><strong>Score:</strong> {rec['score']:.2f}</p>

        </div>

        """
    html_output += "</div>"
    return note_text, html_output

with gr.Blocks() as demo:
    with gr.Tabs():
        with gr.TabItem("Playlist Recommendation"):
            playlist_input = gr.Textbox(
                lines=10,
                label="Assessment Data (JSON)",
                placeholder='''{

  "package": ["Focus", "Insomnia"],

  "report": "Based on your responses, you may struggle with focus, anxiety, and burnout..."

}'''
            )
            playlist_output = gr.HTML(label="Recommended Playlists")
            playlist_btn = gr.Button("Get Playlist Recommendations")
            playlist_btn.click(fn=gradio_recommend_playlist, inputs=playlist_input, outputs=playlist_output)
        with gr.TabItem("Video Recommendation"):
            user_id_input = gr.Textbox(lines=1, label="User ID", placeholder="1")
            note_output = gr.Textbox(label="Recommendation Note", interactive=False)
            videos_output = gr.HTML(label="Recommended Videos")
            videos_btn = gr.Button("Get Video Recommendations")
            videos_btn.click(fn=gradio_recommend_videos, inputs=user_id_input, outputs=[note_output, videos_output])
        with gr.TabItem("User Interaction Ingestion"):
            ingest_input = gr.Textbox(
                lines=10,
                label="User Progress Data (JSON)",
                placeholder='''{

  "user_id": 1,

  "video_id": "abc123",

  "rating": 4.5,

  "time_spent": 300,

  "play_count": 1,

  "completed": false

}'''
            )
            ingest_output = gr.HTML(label="Ingestion Result")
            ingest_btn = gr.Button("Ingest Data")
            ingest_btn.click(fn=gradio_ingest, inputs=ingest_input, outputs=ingest_output)

demo.launch()