hot fix for leaderboard

app.py

@@ -1,6 +1,7 @@
 from io import BytesIO
 from dotenv import load_dotenv
 import os
+from gamification.objects import SimpleIndividualUserLevel
 from utils import *
 from fastapi import FastAPI, File, HTTPException, Header, UploadFile,status
 from tokenManagement import *
@@ -451,14 +452,12 @@ def create_leaderboard_ranking( document: LeaderBoardRanking) -> bool:
 
     
 def get_all_users(user_id =None) -> List:
-
     client = MongoClient(MONGO_URI)
     db = client.crayonics
     collection = db['users']
     # Insert the document
     if user_id==None:
         results= collection.find()
-    
         if results:
             result = [result for result in  results]
             return result
@@ -533,67 +532,201 @@ def handle_change2(new_point):
             logger.error(f"Error adding user {user_id} to leaderboard: {e}")
 
 
+def handle_change3(userId:str):
+    logger.info(f"Extra info: {userId}")
+    print("No leaderboard so creating one now")
+
+    print("========================================================")
+    print("Leveled User", userId)
+
+    if not userId:
+        raise ValueError("User ID not found in inserted document")
+
+                # Fetch user details
+    try:
+        leveleduser = get_all_users(user_id=userId)
+        print("========================================================")
+        print("Leveled User", leveleduser)
+    except Exception as e:
+        logger.error(f"Error fetching user details for userId {userId}: {e}")
+        leveleduser = {}  # Default empty dict to prevent KeyError
+
+        # Fetch user points
+    try:
+        points = get_all_simple_points_func(userId=userId)
+        print("========================================================")
+        print("Points", points)
+    except Exception as e:
+        logger.error(f"Error fetching points for userId {userId}: {e}")
+        points = SimpleIndividualUserLevel(totalpoints=0)
+
+        # Fetch dream job
+    try:
+        tempDreamJob = get_dream_job(userId=userId)
+        print("========================================================")
+        print("Temp dream job", tempDreamJob)
+    except Exception as e:
+        logger.error(f"Error fetching dream job for userId {userId}: {e}")
+        tempDreamJob = None
+
+        dreamJob = tempDreamJob if isinstance(tempDreamJob, str) else "IncompleteProfile"
+        print("DreamJob", dreamJob)
+
+        # Insert into leaderboard
+    try:
+        create_leaderboard_ranking(LeaderBoardRanking(
+                        userId=userId,
+                        firstName=leveleduser.get('first_name', 'Unknown'),
+                        lastName=leveleduser.get('last_name', 'Unknown'),
+                        totalpoints=points.totalpoints,
+                        lastUpdated=datetime.now(),
+                        careerPath=dreamJob,
+                    ))
+    except Exception as e:
+        logger.error(f"Error adding user {userId} to leaderboard: {e}")
+
+    except Exception as e:
+                logger.error(f"Unexpected error processing change document: {e}")
+
+
 # A function to handle changes
-def handle_change(change=None,new_point=None):
-    import logging
-    logging.basicConfig(level=logging.INFO)
-    logger = logging.getLogger(__name__)
-    
-    if new_point!=None:
-        collections = db.list_collection_names()
-        logger.info(f"Extra info: {new_point}")
+def handle_change(change=None):
+
+    print("Change detected in point making changes immediately")
+    logger.info(f"Change detected: {change}")
+    # add everybodies points and add it to the leaderboard table
+    collections = db.list_collection_names()
+    if "LeaderBoard" not in collections:
         print("No leaderboard so creating one now")
         users = get_all_users()
         for user in users:
             print("inserting user",f"user id {user['_id']}")
             points = get_all_simple_points_func(userId=str(user['_id']))
             tempDreamJob =  get_dream_job(userId=str(user['_id'])) 
+            
             dreamJob = tempDreamJob if type(tempDreamJob)==str else "IncompleteProfile"
+            
             create_leaderboard_ranking(LeaderBoardRanking(userId=str(user['_id']),firstName=user['first_name'],lastName=user['last_name'],totalpoints=points.totalpoints,lastUpdated=datetime.now(),careerPath=dreamJob,))
-      
-                
-        
-    elif new_point==None and change!=None:
-        print("Change detected in point making changes immediately")
-        logger.info(f"Change detected: {change}")
-        # add everybodies points and add it to the leaderboard table
-        collections = db.list_collection_names()
-        if "LeaderBoard" not in collections:
-            print("No leaderboard so creating one now")
-            users = get_all_users()
-            for user in users:
-                print("inserting user",f"user id {user['_id']}")
-                points = get_all_simple_points_func(userId=str(user['_id']))
-                tempDreamJob =  get_dream_job(userId=str(user['_id'])) 
-                
-                dreamJob = tempDreamJob if type(tempDreamJob)==str else "IncompleteProfile"
-                
-                create_leaderboard_ranking(LeaderBoardRanking(userId=str(user['_id']),firstName=user['first_name'],lastName=user['last_name'],totalpoints=points.totalpoints,lastUpdated=datetime.now(),careerPath=dreamJob,))
 
-        else:
-            if change['operationType'] == 'insert':
+    else:
+        if change['operationType'] == 'insert':
+            try:
                 # Extract the full document
-                full_document = change['fullDocument']
-                
-                # Extract the userId and numOfPoints
-                
-                user_id =full_document.get('userId')
-                leveleduser = get_all_users(userId=user_id)
-                points = get_all_simple_points_func(userId=user_id)
-                tempDreamJob =  get_dream_job(userId=user_id) 
-                dreamJob = tempDreamJob if type(tempDreamJob)==str else "IncompleteProfile"
-                create_leaderboard_ranking(LeaderBoardRanking(userId=user_id,firstName=leveleduser['first_name'],lastName=leveleduser['last_name'],totalpoints=points.totalpoints,lastUpdated=datetime.now(),careerPath=dreamJob,))
-            elif change['operationType'] == 'update':
-                dockey = str(change['documentKey']['_id'])
-                user_id = get_user_id_from_docKey(dockId=dockey)
-                leveleduser = get_all_users(user_id=user_id)
-                points = get_all_simple_points_func(userId=user_id)
-                tempDreamJob =  get_dream_job(userId=user_id)
-                dreamJob = tempDreamJob if type(tempDreamJob)==str else "IncompleteProfile"
-                create_leaderboard_ranking(LeaderBoardRanking(userId=user_id,firstName=leveleduser['first_name'],lastName=leveleduser['last_name'],totalpoints=points.totalpoints,lastUpdated=datetime.now(),careerPath=dreamJob,))
+                full_document = change.get('fullDocument', {})
+                user_id = full_document.get('userId')
+
+                print("========================================================")
+                print("Leveled User", user_id)
+
+                if not user_id:
+                    raise ValueError("User ID not found in inserted document")
+
+                # Fetch user details
+                try:
+                    leveleduser = get_all_users(user_id=user_id)
+                    print("========================================================")
+                    print("Leveled User", leveleduser)
+                except Exception as e:
+                    logger.error(f"Error fetching user details for userId {user_id}: {e}")
+                    leveleduser = {}  # Default empty dict to prevent KeyError
+
+                # Fetch user points
+                try:
+                    points = get_all_simple_points_func(userId=user_id)
+                    print("========================================================")
+                    print("Points", points)
+                except Exception as e:
+                    logger.error(f"Error fetching points for userId {user_id}: {e}")
+                    points = SimpleIndividualUserLevel(totalpoints=0)
+
+                # Fetch dream job
+                try:
+                    tempDreamJob = get_dream_job(userId=user_id)
+                    print("========================================================")
+                    print("Temp dream job", tempDreamJob)
+                except Exception as e:
+                    logger.error(f"Error fetching dream job for userId {user_id}: {e}")
+                    tempDreamJob = None
+
+                dreamJob = tempDreamJob if isinstance(tempDreamJob, str) else "IncompleteProfile"
+                print("DreamJob", dreamJob)
+
+                # Insert into leaderboard
+                try:
+                    create_leaderboard_ranking(LeaderBoardRanking(
+                        userId=user_id,
+                        firstName=leveleduser.get('first_name', 'Unknown'),
+                        lastName=leveleduser.get('last_name', 'Unknown'),
+                        totalpoints=points.totalpoints,
+                        lastUpdated=datetime.now(),
+                        careerPath=dreamJob,
+                    ))
+                except Exception as e:
+                    logger.error(f"Error adding user {user_id} to leaderboard: {e}")
+
+            except Exception as e:
+                logger.error(f"Unexpected error processing change document: {e}")
+        elif change['operationType'] == 'update':
+            try:
+                # Extract the full document
+                doc_id = str( change.get('documentKey')['_id'])
+
+                print("========================================================")
+                print("document Id", doc_id)
+                user_id = get_user_id_from_docKey(doc_id)
+
+                if not user_id:
+                    raise ValueError("User ID not found in inserted document")
+
+                # Fetch user details
+                try:
+                    leveleduser = get_all_users(user_id=user_id)
+                    print("========================================================")
+                    print("Leveled User", leveleduser)
+                except Exception as e:
+                    logger.error(f"Error fetching user details for userId {user_id}: {e}")
+                    leveleduser = {}  # Default empty dict to prevent KeyError
+
+                # Fetch user points
+                try:
+                    points = get_all_simple_points_func(userId=user_id)
+                    print("========================================================")
+                    print("Points", points)
+                except Exception as e:
+                    logger.error(f"Error fetching points for userId {user_id}: {e}")
+                    points = SimpleIndividualUserLevel(totalpoints=0)
+
+                # Fetch dream job
+                try:
+                    tempDreamJob = get_dream_job(userId=user_id)
+                    print("========================================================")
+                    print("Temp dream job", tempDreamJob)
+                except Exception as e:
+                    logger.error(f"Error fetching dream job for userId {user_id}: {e}")
+                    tempDreamJob = None
+
+                dreamJob = tempDreamJob if isinstance(tempDreamJob, str) else "IncompleteProfile"
+                print("DreamJob", dreamJob)
+
+                # Insert into leaderboard
+                try:
+                    create_leaderboard_ranking(LeaderBoardRanking(
+                        userId=user_id,
+                        firstName=leveleduser.get('first_name', 'Unknown'),
+                        lastName=leveleduser.get('last_name', 'Unknown'),
+                        totalpoints=points.totalpoints,
+                        lastUpdated=datetime.now(),
+                        careerPath=dreamJob,
+                    ))
+                except Exception as e:
+                    logger.error(f"Error adding user {user_id} to leaderboard: {e}")
+
+            except Exception as e:
+                logger.error(f"Unexpected error processing change document: {e}")
+
+
                 
-            
-        logger.info(f"Change detected: {dumps(change)}")
+    logger.info(f"Change detected:")
 
 # Function to run the change stream in a separate thread (non-blocking)
 def watch_change_stream():
@@ -606,3 +739,4 @@ def watch_change_stream():
 @app.on_event("startup")
 def start_change_stream():
     threading.Thread(target=watch_change_stream, daemon=True).start()
+

gamification/pointLogic.py

@@ -3,7 +3,8 @@
 from gamification.objects import UserPoints ,SimpleIndividualUserLevel,IndividualUserLevel,UserLevel
 from gamification.imports import *
 from gamification.levelLogic import get_all_levels_func
-
+from concurrent.futures import ThreadPoolExecutor
+executor = ThreadPoolExecutor(max_workers=5)
 # utils 
 def get_particular_level(totalPoints,dreamJob)->UserLevel:
     # query db and get the results of all the level probably re use a function 
@@ -29,11 +30,13 @@ def get_dream_job(userId):
 
 
 def create_points_func(document:UserPoints)->bool:
+    from app import handle_change3
     db_uri = MONGO_URI
     db_name = "crayonics"
     collection_name="Points"
     client = MongoClient(db_uri)
     db = client[db_name]
+    
     collection = db[collection_name]
     # Insert the document
     
@@ -41,7 +44,7 @@ def create_points_func(document:UserPoints)->bool:
         doc = document.model_dump()
         doc['earnedAt']=datetime.now()
         result = collection.insert_one(doc)
-        
+        executor.submit (handle_change3 , document.userId)
         return True
     else:
         client.close()

gamification/routes.py

@@ -132,10 +132,7 @@ def get_leaderboard(background_tasks: BackgroundTasks)->List[Ranker]:
         list_of_rankers = []
         result = get_top_30()
         background_tasks.add_task(handle_change2, 2)
-        # executor.submit(handle_change2,2)
-
         list_of_rankers = [Ranker(**ranker) for ranker in result]
-        # handle_change(new_point="userId")
         return list_of_rankers
     except Exception as e:
         raise HTTPException(status_code=500,detail=f"{e}")

testing.py

@@ -0,0 +1,28 @@
+
+from typing import List
+
+from bson import ObjectId
+from pymongo import MongoClient
+
+from app import MONGO_URI
+
+
+def get_all_users(user_id:str =None) -> List:
+    client = MongoClient(MONGO_URI)
+    db = client.crayonics
+    collection = db['users']
+    # Insert the document
+    if user_id==None:
+        results= collection.find()
+        if results:
+            result = [result for result in  results]
+            return result
+        
+        client.close()
+    else:
+        result = collection.find_one(filter={"_id":ObjectId(user_id.strip())})
+        return result
+
+
+
+print(get_all_users(user_id="67c9b68678fbf39f4ed94e01"))