Spaces:
Sleeping
Sleeping
| 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") | |
| supabase_client = create_client(SUPABASE_URL, SUPABASE_KEY) | |
| pinecone_api_key = os.getenv("PINECONE") | |
| 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"<div style='color: red;'>Error parsing input: {e}</div>" | |
| res = ingest_user_progress(supabase_client, user_id, video_id, rating, time_spent, play_count, completed) | |
| return f"<div style='color: green;'>Ingested data: {res}</div>" | |
| 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 "<div style='color: red;'>Error: Invalid JSON format</div>" | |
| if "package" not in assessment_data or "report" not in assessment_data: | |
| return "<div style='color: red;'>Error: Missing 'package' or 'report' field</div>" | |
| recs = recommend_playlists_by_package_and_module(assessment_data, index, model) | |
| html_output = """ | |
| <div style=" | |
| display: flex; | |
| flex-direction: column; | |
| gap: 30px; | |
| padding: 20px; | |
| font-family: Arial, sans-serif; | |
| "> | |
| """ | |
| for pkg, mod_recs in recs.items(): | |
| html_output += f"<h2 style='color: #2d3436;'>{pkg} Package</h2>" | |
| html_output += "<div style='display: flex; flex-wrap: wrap; gap: 20px;'>" | |
| for mod, rec in mod_recs.items(): | |
| card_html = f""" | |
| <div style=" | |
| border: 1px solid #e0e0e0; | |
| border-radius: 10px; | |
| padding: 20px; | |
| background: white; | |
| width: 300px; | |
| box-shadow: 0 2px 4px rgba(0,0,0,0.1); | |
| "> | |
| <h3 style="margin: 0 0 12px 0; color: #0984e3;">{mod} Module</h3> | |
| <h4 style="margin: 0 0 8px 0; color: #2d3436;">{rec['title']}</h4> | |
| <p style="margin: 0; color: #636e72; line-height: 1.5;">{rec['description']}</p> | |
| </div> | |
| """ | |
| html_output += card_html | |
| html_output += "</div>" | |
| html_output += "</div>" | |
| return html_output | |
| def recommend_videos(user_id: int, K: int = 5, M: int = 10, N: int = 5) -> List[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 [] | |
| 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'] | |
| engagement = rating + 2 * completed_val + play_count | |
| inter['engagement'] = engagement | |
| 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']} | |
| 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'] | |
| recommendations.append({ | |
| 'video_id': vid, | |
| 'title': metadata.get('video_title'), | |
| 'description': metadata.get('video_description'), | |
| 'score': data['total_score'] | |
| }) | |
| return recommendations | |
| def gradio_recommend_videos(user_id_input): | |
| try: | |
| user_id = int(user_id_input) | |
| except Exception as e: | |
| return f"<div style='color: red;'>Error: {e}</div>" | |
| recs = recommend_videos(user_id) | |
| if not recs: | |
| return "<div style='color: #636e72;'>No video recommendations found for this user.</div>" | |
| html_output = """ | |
| <div style=" | |
| display: flex; | |
| flex-direction: column; | |
| gap: 30px; | |
| padding: 20px; | |
| font-family: Arial, sans-serif; | |
| "> | |
| """ | |
| html_output += "<h2 style='color: #2d3436;'>Recommended Videos</h2>" | |
| html_output += "<div style='display: flex; flex-wrap: wrap; gap: 20px;'>" | |
| for rec in recs: | |
| card_html = f""" | |
| <div style=" | |
| border: 1px solid #e0e0e0; | |
| border-radius: 10px; | |
| padding: 20px; | |
| background: white; | |
| width: 300px; | |
| box-shadow: 0 2px 4px rgba(0,0,0,0.1); | |
| "> | |
| <h3 style="margin: 0 0 12px 0; color: #0984e3;">{rec['title']}</h3> | |
| <p style="margin: 0 0 8px 0; color: #636e72;">{rec['description']}</p> | |
| <p style="margin: 0; color: #2d3436;">Score: {rec['score']:.2f}</p> | |
| </div> | |
| """ | |
| html_output += card_html | |
| html_output += "</div></div>" | |
| return 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. The Focus and Insomnia packages can help improve your mental clarity and sleep quality." | |
| }''' | |
| ) | |
| playlist_output = gr.HTML(label="Recommended Playlists") | |
| playlist_btn = gr.Button("Get Playlist Recommendations") | |
| playlist_btn.click(gradio_recommend_playlist, playlist_input, playlist_output) | |
| with gr.TabItem("Video Recommendation"): | |
| user_id_input = gr.Textbox(lines=1, label="User ID", placeholder="1") | |
| videos_output = gr.HTML(label="Recommended Videos") | |
| videos_btn = gr.Button("Get Video Recommendations") | |
| videos_btn.click(gradio_recommend_videos, user_id_input, 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(gradio_ingest, ingest_input, ingest_output) | |
| demo.launch() | |