Upload 2 files

app.py

@@ -0,0 +1,167 @@
+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")
+iface.launch()

requirements.txt

@@ -0,0 +1,5 @@
+gradio
+faiss-cpu
+sentence-transformers
+pymongo
+google-generativeai