from app import handle_change from gamification.objects import UserPoints ,SimpleIndividualUserLevel,IndividualUserLevel,UserLevel from gamification.imports import * from gamification.levelLogic import get_all_levels_func # utils def get_particular_level(totalPoints,dreamJob)->UserLevel: # query db and get the results of all the level probably re use a function levels = get_all_levels_func() particularLevel= [level for level in levels if (level.maxPoints >= totalPoints) and (level.minPoints<=totalPoints)and (level.careerPath==dreamJob)] defaulLevel= [level for level in levels if level.levelName=="default" ] if len(particularLevel)>0: return particularLevel return defaulLevel def get_dream_job(userId): db_uri=MONGO_URI db_name = "crayonics" collection_name = "Questionaire" client = MongoClient(db_uri) db = client[db_name] collection = db[collection_name] questionaire_doc = collection.find_one({"userId": userId}) if questionaire_doc: return questionaire_doc['dreamRole'] else: return False def create_points_func(document:UserPoints)->bool: 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 if document!=None: doc = document.model_dump() doc['earnedAt']=datetime.now() result = collection.insert_one(doc) handle_change(new_point={"userId",document.userId}) return True else: client.close() return False def get_all_points_func(userId) -> IndividualUserLevel: # MongoDB URI and configuration db_uri = MONGO_URI db_name = "crayonics" collection_name = "Points" client = MongoClient(db_uri) db = client[db_name] collection = db[collection_name] dreamJob = get_dream_job(userId=userId) # Fetch all documents from the collection point_cursor = collection.find({"userId": userId}) # This returns a cursor to the documents # Convert the cursor to a list so we can reuse it points_list = list(point_cursor) # Calculate the total points totalPoints = sum([point['numOfPoints'] for point in points_list]) particularLevelInfo = get_particular_level(dreamJob=dreamJob,totalPoints=totalPoints) # Create the individual points list individualPoints = [indpoint for indpoint in points_list] print([pointss for pointss in individualPoints if all(val is not None for val in pointss.values())]) # Create the IndividualUserLevel object with totalPoints and individualPoints points = IndividualUserLevel(totalpoints=totalPoints,levelName=particularLevelInfo[0].levelName,minPoints=particularLevelInfo[0].minPoints,maxPoints=particularLevelInfo[0].maxPoints, individualPoints=individualPoints) return points def get_all_simple_points_func(userId) -> SimpleIndividualUserLevel: # MongoDB URI and configuration db_uri = MONGO_URI db_name = "crayonics" collection_name = "Points" client = MongoClient(db_uri) db = client[db_name] collection = db[collection_name] dreamJob = get_dream_job(userId=userId) # Fetch all documents from the collection point_cursor = collection.find({"userId": userId}) # This returns a cursor to the documents # Convert the cursor to a list so we can reuse it points_list = list(point_cursor) # Calculate the total points totalPoints = sum([point['numOfPoints'] for point in points_list]) particularLevelInfo = get_particular_level(dreamJob=dreamJob,totalPoints=totalPoints) # Create the individual points list # Create the IndividualUserLevel object with totalPoints and individualPoints points = SimpleIndividualUserLevel(totalpoints=totalPoints,levelName=particularLevelInfo[0].levelName,minPoints=particularLevelInfo[0].minPoints,maxPoints=particularLevelInfo[0].maxPoints) return points