Update app.py

app.py

@@ -1,167 +1,168 @@
-import gradio as gr
-import google.generativeai as genai
-import faiss
-import numpy as np
-from sentence_transformers import SentenceTransformer
-from pymongo import MongoClient
-
-# ✅ Configure Gemini API
-genai.configure(api_key="AIzaSyBWMLGBoKDeA7_Z_AzHDWtFBKOJ91BJnaY")
-
-# ✅ Load Sentence Transformer Model
-model = SentenceTransformer('all-MiniLM-L6-v2')
-
-# ✅ Connect to MongoDB
-MONGO_URI = "mongodb+srv://aiworkspaceadku:FstomozFvaR6maVs@cluster0.5dtl1.mongodb.net/AiWork"
-client = MongoClient(MONGO_URI)
-db = client["AiWork"]
-
-# ✅ Fetch latest data dynamically
-def fetch_latest_data():
-    return {
-        "users": list(db.users.find()),
-        "teams": list(db.teams.find()),
-        "projects": list(db.projects.find()),
-        "modules": list(db.modules.find()),
-        "documents": list(db.documents.find()),
-        "schedules": list(db.schedules.find())
-    }
-
-def generate_sentences(db):
-    users, teams, projects, modules, documents, schedules = (
-        db["users"], db["teams"], db["projects"], db["modules"], db["documents"], db["schedules"]
-    )
-    user_sentences = {}  # Store categorized sentences per user
-    
-    for user in users:
-        username = user.get("username", "Unknown User")
-        email = user.get("email", "Unknown Email")
-        
-        if email not in user_sentences:
-            user_sentences[email] = {
-                "Teams": [],
-                "Projects": [],
-                "Modules & Tasks": [],
-                "Documents": [],
-                "Schedules": []
-            }
-        
-        # User team ownership and membership
-        owned_teams = [team for team in teams if team.get("owner", {}).get("email") == email]
-        if owned_teams:
-            team_names = ", ".join(f'"{team["teamName"]}"' for team in owned_teams)
-            user_sentences[email]["Teams"].append(f"User {username} owns the teams: {team_names}.")
-        
-        member_teams = [team for team in teams if any(m["email"] == email for m in team.get("members", []))]
-        if member_teams:
-            team_names = ", ".join(f'"{team["teamName"]}"' for team in member_teams)
-            user_sentences[email]["Teams"].append(f"User {username} is a member of the teams: {team_names}.")
-        
-        # Find projects in teams they own or are part of
-        relevant_teams = owned_teams + member_teams
-        team_ids = [str(team["_id"]) for team in relevant_teams]
-        user_projects = [p for p in projects if str(p.get("owner", {}).get("teamId")) in team_ids]
-        
-        if user_projects:
-            for project in user_projects:
-                proj_name = project["projName"]
-                user_sentences[email]["Projects"].append(f"User {username} is involved in project {proj_name}.")
-                
-                # Find modules under this project
-                proj_modules = [m for m in modules if str(m.get("projId")) == str(project["_id"])]
-                if proj_modules:
-                    for module in proj_modules:
-                        module_name = module["moduleName"]
-                        user_sentences[email]["Modules & Tasks"].append(f"In project {proj_name}, module {module_name} exists.")
-                        
-                        # Find tasks in this module assigned to the user
-                        assigned_tasks = [
-                            task for task in module.get("tasks", [])
-                            if any(a["email"] == email for a in task.get("assignedTo", []))
-                        ]
-                        if assigned_tasks:
-                            task_names = ", ".join(f'"{t["taskName"]}"' for t in assigned_tasks)
-                            user_sentences[email]["Modules & Tasks"].append(f"Tasks assigned to {username} in {module_name}: {task_names}.")
-                
-                # Find documents in this project
-                proj_docs = [d for d in documents if str(d.get("owner", {}).get("projId")) == str(project["_id"])]
-                if proj_docs:
-                    doc_names = ", ".join(f'"{d["title"]}"' for d in proj_docs)
-                    user_sentences[email]["Documents"].append(f"Documents related to project {proj_name}: {doc_names}.")
-        
-        # Find meeting schedules related to their teams
-        user_schedules = [s for s in schedules if str(s.get("teamId")) in team_ids]
-        if user_schedules:
-            for schedule in user_schedules:
-                schedule_detail = f'{schedule["moto"]} scheduled on {schedule["date"]} at {schedule["time"]}.'
-                user_sentences[email]["Schedules"].append(schedule_detail)
-    
-    return user_sentences
-
-# ✅ Update FAISS Index dynamically
-def update_faiss_index(user_sentences):
-    faiss_indices = {}  
-    for email, categories in user_sentences.items():
-        sentences = sum(categories.values(), [])  
-        
-        if not sentences:
-            continue
-        
-        embeddings = model.encode(sentences, convert_to_numpy=True)
-        embedding_dim = embeddings.shape[1]
-        
-        index = faiss.IndexFlatL2(embedding_dim)  
-        index.add(embeddings)  
-        
-        faiss_indices[email] = {"index": index, "sentences": sentences}  
-
-    return faiss_indices
-
-# ✅ Query FAISS Index
-def get_relevant_sentences(email, query, faiss_indices):
-    if email not in faiss_indices:
-        return ["User not found or no data available."]
-    
-    index_data = faiss_indices[email]
-    index = index_data["index"]
-    sentences = index_data["sentences"]
-
-    # Compute query embedding
-    query_embedding = model.encode([query], convert_to_numpy=True)
-    k = 100
-    distances, indices = index.search(query_embedding, k)
-
-    # Filter sentences based on FAISS similarity threshold
-    threshold = 1.5
-    filtered_sentences = [sentences[idx] for dist, idx in zip(distances[0], indices[0]) if dist < threshold]
-
-    return filtered_sentences if filtered_sentences else ["No relevant information found."]
-
-# ✅ Generate response using Gemini API
-def generate_response(email, query):
-    filtered_sentences = get_relevant_sentences(email, query, faiss_indices)
-    
-    if filtered_sentences == ["No relevant information found."]:
-        return "No relevant information found."
-    
-    prompt = f"Based on query '{query}', generate a short answer using:\n\n" + "\n".join(filtered_sentences)
-    model = genai.GenerativeModel("gemini-1.5-flash")
-    response = model.generate_content(prompt)
-    
-    return response.text
-
-# ✅ Real-time data update function
-def real_time_update():
-    latest_data = fetch_latest_data()  
-    user_sentences = generate_sentences(latest_data)  
-    return update_faiss_index(user_sentences)
-
-# ✅ Run initial update
-faiss_indices = real_time_update()
-
-# ✅ Gradio Web UI
-def chat(email, query):
-    return generate_response(email, query)
-
-iface = gr.Interface(fn=chat, inputs=["text", "text"], outputs="text", title="AI Workspace Assistant")
+import os
+import gradio as gr
+import google.generativeai as genai
+import faiss
+import numpy as np
+from sentence_transformers import SentenceTransformer
+from pymongo import MongoClient
+
+# ✅ Configure Gemini API
+genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
+
+# ✅ Load Sentence Transformer Model
+model = SentenceTransformer('all-MiniLM-L6-v2')
+
+# ✅ Connect to MongoDB
+MONGO_URI = os.getenv("MONGO_URI")
+client = MongoClient(MONGO_URI)
+db = client["AiWork"]
+
+# ✅ Fetch latest data dynamically
+def fetch_latest_data():
+    return {
+        "users": list(db.users.find()),
+        "teams": list(db.teams.find()),
+        "projects": list(db.projects.find()),
+        "modules": list(db.modules.find()),
+        "documents": list(db.documents.find()),
+        "schedules": list(db.schedules.find())
+    }
+
+def generate_sentences(db):
+    users, teams, projects, modules, documents, schedules = (
+        db["users"], db["teams"], db["projects"], db["modules"], db["documents"], db["schedules"]
+    )
+    user_sentences = {}  # Store categorized sentences per user
+    
+    for user in users:
+        username = user.get("username", "Unknown User")
+        email = user.get("email", "Unknown Email")
+        
+        if email not in user_sentences:
+            user_sentences[email] = {
+                "Teams": [],
+                "Projects": [],
+                "Modules & Tasks": [],
+                "Documents": [],
+                "Schedules": []
+            }
+        
+        # User team ownership and membership
+        owned_teams = [team for team in teams if team.get("owner", {}).get("email") == email]
+        if owned_teams:
+            team_names = ", ".join(f'"{team["teamName"]}"' for team in owned_teams)
+            user_sentences[email]["Teams"].append(f"User {username} owns the teams: {team_names}.")
+        
+        member_teams = [team for team in teams if any(m["email"] == email for m in team.get("members", []))]
+        if member_teams:
+            team_names = ", ".join(f'"{team["teamName"]}"' for team in member_teams)
+            user_sentences[email]["Teams"].append(f"User {username} is a member of the teams: {team_names}.")
+        
+        # Find projects in teams they own or are part of
+        relevant_teams = owned_teams + member_teams
+        team_ids = [str(team["_id"]) for team in relevant_teams]
+        user_projects = [p for p in projects if str(p.get("owner", {}).get("teamId")) in team_ids]
+        
+        if user_projects:
+            for project in user_projects:
+                proj_name = project["projName"]
+                user_sentences[email]["Projects"].append(f"User {username} is involved in project {proj_name}.")
+                
+                # Find modules under this project
+                proj_modules = [m for m in modules if str(m.get("projId")) == str(project["_id"])]
+                if proj_modules:
+                    for module in proj_modules:
+                        module_name = module["moduleName"]
+                        user_sentences[email]["Modules & Tasks"].append(f"In project {proj_name}, module {module_name} exists.")
+                        
+                        # Find tasks in this module assigned to the user
+                        assigned_tasks = [
+                            task for task in module.get("tasks", [])
+                            if any(a["email"] == email for a in task.get("assignedTo", []))
+                        ]
+                        if assigned_tasks:
+                            task_names = ", ".join(f'"{t["taskName"]}"' for t in assigned_tasks)
+                            user_sentences[email]["Modules & Tasks"].append(f"Tasks assigned to {username} in {module_name}: {task_names}.")
+                
+                # Find documents in this project
+                proj_docs = [d for d in documents if str(d.get("owner", {}).get("projId")) == str(project["_id"])]
+                if proj_docs:
+                    doc_names = ", ".join(f'"{d["title"]}"' for d in proj_docs)
+                    user_sentences[email]["Documents"].append(f"Documents related to project {proj_name}: {doc_names}.")
+        
+        # Find meeting schedules related to their teams
+        user_schedules = [s for s in schedules if str(s.get("teamId")) in team_ids]
+        if user_schedules:
+            for schedule in user_schedules:
+                schedule_detail = f'{schedule["moto"]} scheduled on {schedule["date"]} at {schedule["time"]}.'
+                user_sentences[email]["Schedules"].append(schedule_detail)
+    
+    return user_sentences
+
+# ✅ Update FAISS Index dynamically
+def update_faiss_index(user_sentences):
+    faiss_indices = {}  
+    for email, categories in user_sentences.items():
+        sentences = sum(categories.values(), [])  
+        
+        if not sentences:
+            continue
+        
+        embeddings = model.encode(sentences, convert_to_numpy=True)
+        embedding_dim = embeddings.shape[1]
+        
+        index = faiss.IndexFlatL2(embedding_dim)  
+        index.add(embeddings)  
+        
+        faiss_indices[email] = {"index": index, "sentences": sentences}  
+
+    return faiss_indices
+
+# ✅ Query FAISS Index
+def get_relevant_sentences(email, query, faiss_indices):
+    if email not in faiss_indices:
+        return ["User not found or no data available."]
+    
+    index_data = faiss_indices[email]
+    index = index_data["index"]
+    sentences = index_data["sentences"]
+
+    # Compute query embedding
+    query_embedding = model.encode([query], convert_to_numpy=True)
+    k = 100
+    distances, indices = index.search(query_embedding, k)
+
+    # Filter sentences based on FAISS similarity threshold
+    threshold = 1.5
+    filtered_sentences = [sentences[idx] for dist, idx in zip(distances[0], indices[0]) if dist < threshold]
+
+    return filtered_sentences if filtered_sentences else ["No relevant information found."]
+
+# ✅ Generate response using Gemini API
+def generate_response(email, query):
+    filtered_sentences = get_relevant_sentences(email, query, faiss_indices)
+    
+    if filtered_sentences == ["No relevant information found."]:
+        return "No relevant information found."
+    
+    prompt = f"Based on query '{query}', generate a short answer using:\n\n" + "\n".join(filtered_sentences)
+    model = genai.GenerativeModel("gemini-1.5-flash")
+    response = model.generate_content(prompt)
+    
+    return response.text
+
+# ✅ Real-time data update function
+def real_time_update():
+    latest_data = fetch_latest_data()  
+    user_sentences = generate_sentences(latest_data)  
+    return update_faiss_index(user_sentences)
+
+# ✅ Run initial update
+faiss_indices = real_time_update()
+
+# ✅ Gradio Web UI
+def chat(email, query):
+    return generate_response(email, query)
+
+iface = gr.Interface(fn=chat, inputs=["text", "text"], outputs="text", title="AI Workspace Assistant")
 iface.launch()