from datetime import datetime from pydantic import model_validator, BaseModel from typing import List, Optional, Union from bson import ObjectId class UserLevel(BaseModel): _id: Optional[ObjectId]=None # Make sure _id can be Optional id:Optional[str]=None maxPoints: int minPoints: int levelName: str careerPath: str levelNumber: int # To convert MongoDB ObjectId to string class Config: json_encoders = { ObjectId: str } # Custom validator to handle the ObjectId conversion if needed @model_validator(mode='before') def handle_objectid(cls, values): if '_id' in values and isinstance(values['_id'], ObjectId): values['id'] = str(values['_id']) # Convert ObjectId to string return values class UserPoints(BaseModel): pointsId:str userId:str class Config: json_encoders = { ObjectId: str } class LearningProgress(BaseModel): course_completion: Optional[int] = None # Points for course completion module_completion: Optional[int] = None # Points for module completion daily_learning_streak: Optional[int] = None # Points for daily learning streak quiz_assessment_completion: Optional[int] = None# Points for quiz/assessment completion passing_score: Optional[int] = None # Points for passing assessments class Config: json_encoders = { ObjectId: str } class SkillDevelopment(BaseModel): skill_milestone_achievements: Optional[int] = None # Points for achieving milestones cross_discipline_learning: Optional[int] = None # Points for completing courses in different tech categories practical_project_submission: Optional[int] = None # Points for submitting a practical project project_peer_review: Optional[int] = None # Points for reviewing a project class Config: json_encoders = { ObjectId: str } class PlatformEngagement(BaseModel): profile_completion: Optional[int] = None # Points for completing profile connecting_learning_accounts: Optional[int] = None # Points for connecting learning accounts daily_check_in: Optional[int] = None # Points for daily check-in community_participation: Optional[int] = None # Points for community participation providing_feedback: Optional[int] = None # Points for providing feedback class Config: json_encoders = { ObjectId: str } class PointMultipliersWeekendWarrior(BaseModel): weekendWarrior: float = 1.5 class Config: json_encoders = { ObjectId: str } class PointMultipliersFocusMode(BaseModel): focusMode: float = 2 class Config: json_encoders = { ObjectId: str } class PointMultipliersSkillGapTargeting(BaseModel): skillGapTargeting: float = 2 class Config: json_encoders = { ObjectId: str } class PointMultipliersTrendingTech(BaseModel): trendingTech: float = 1.5 class Config: json_encoders = { ObjectId: str } class Points(BaseModel): userId: str numOfPoints: Optional[float] = None # Initially set to None learningProgress: Optional[LearningProgress] = None skillDevelopment: Optional[SkillDevelopment] = None platformEngagement: Optional[PlatformEngagement] = None weekendWarrior: Optional[PointMultipliersWeekendWarrior] = None trendingTech: Optional[PointMultipliersTrendingTech]=None skillGapTargeting:Optional[PointMultipliersSkillGapTargeting]=None focusMode:Optional[PointMultipliersFocusMode]=None earnedAt:Optional[datetime]=datetime.now() @model_validator(mode='before') def calculate_num_of_points(cls, values): total_points = 0 # Add points from LearningProgress learning_progress = values.get('learningProgress') if learning_progress: total_points += sum(filter(None, [ learning_progress.course_completion, learning_progress.module_completion, learning_progress.daily_learning_streak, learning_progress.quiz_assessment_completion, learning_progress.passing_score ])) # Add points from SkillDevelopment skill_development = values.get('skillDevelopment') if skill_development: total_points += sum(filter(None, [ skill_development.skill_milestone_achievements, skill_development.cross_discipline_learning, skill_development.practical_project_submission, skill_development.project_peer_review ])) # Add points from PlatformEngagement platform_engagement = values.get('platformEngagement') if platform_engagement: total_points += sum(filter(None, [ platform_engagement.profile_completion, platform_engagement.connecting_learning_accounts, platform_engagement.daily_check_in, platform_engagement.community_participation, platform_engagement.providing_feedback ])) # Apply multipliers if present PointMultipliersWeekendWarrior = values.get('weekendWarrior') if PointMultipliersWeekendWarrior: total_points *= sum([ PointMultipliersWeekendWarrior.weekendWarrior ]) PointMultipliersTrendingTech = values.get('trendingTech') if PointMultipliersTrendingTech: total_points *= sum([ PointMultipliersTrendingTech.trendingTech ]) PointMultipliersSkillGapTargeting = values.get('skillGapTargeting') if PointMultipliersSkillGapTargeting: total_points *= sum([ PointMultipliersSkillGapTargeting.skillGapTargeting ]) PointMultipliersFocusMode = values.get('focusMode') if PointMultipliersFocusMode: total_points *= sum([ PointMultipliersFocusMode.focusMode, ]) values['numOfPoints'] = total_points # Store calculated points return values class Config: json_encoders = { ObjectId: str } class UserFeedback(BaseModel): userId:str feedback:str rating:int respondedAt:datetime class Config: json_encoders = { ObjectId: str } class Feedback(BaseModel): feedback:str rating:int class Config: json_encoders = { ObjectId: str } class Customer(BaseModel): email:str first_name:str last_name:str date_Joined: datetime class Config: json_encoders = { ObjectId: str } class CustomerInfo(BaseModel): totalCustomers:int GrowthRate: float GrowthRateType:str daysAgo:int class Config: json_encoders = { ObjectId: str } class individualPoints(BaseModel): numOfPoints:float earnedAt:datetime platformEngagement:Optional[dict]=None learningProgress:Optional[dict]=None skillDevelopment:Optional[dict]=None weekendWarrior:Optional[dict]=None trendingTech:Optional[dict]=None skillGapTargeting:Optional[dict]=None focusMode:Optional[dict]=None starUrl:Optional[str]="https://iili.io/37bdMPe.png" @model_validator(mode='after') def change_star_url(cls,values): weekendWarriorPM = values.weekendWarrior trendingTechPM=values.trendingTech skillGapTargetingPM=values.skillGapTargeting focusModePM=values.focusMode a=0 if weekendWarriorPM !=None: a=a+1 if trendingTechPM !=None: a=a+1 if skillGapTargetingPM !=None: a=a+1 if focusModePM!=None: a=a+1 if a==1 or a==2: values.starUrl='https://iili.io/37b2RWP.png' elif a==3 or a==4: values.starUrl='https://iili.io/37b3Yj2.png' return values class Config: json_encoders={ ObjectId:str } class IndividualUserLevel(BaseModel): totalpoints:float levelName:str levelNumber:int maxPoints:float minPoints:float individualPoints:List[individualPoints] class Config: json_encoders = { ObjectId: str } class SimpleIndividualUserLevel(BaseModel): totalpoints:float levelName:Optional[str]="default" maxPoints:Optional[float]=0 minPoints:Optional[float]=999999999999999999999 levelNumber:Optional[int]=1 class Config: json_encoders = { ObjectId: str } class Ranker(BaseModel): firstName:str lastName:str rank:int totalPoints:Optional[float]=None dreamJob:Optional[str]=None badgeUrl:Optional[str]=None class Config: json_encoders = { ObjectId:str } @model_validator(mode='before') def changeValueNames(cls,values): values['totalPoints']= values.get("totalpoints") values['dreamJob']=values.get("careerPath") if values.get("totalpoints") >=0 and values.get("totalpoints") <=200: values['badgeUrl'] ="https://res.cloudinary.com/dfmzougki/image/upload/v1742757994/Frame_121_LE_upscale_magic_x4_strength_20_similarity_50_1-removebg-preview_qnbmf3.png" elif values.get("totalpoints") >=201 and values.get("totalpoints") <=400: values['badgeUrl'] ="https://res.cloudinary.com/dfmzougki/image/upload/v1742757797/image-21_LE_upscale_magic_x4_strength_20_similarity_50_tone_enhance_30_color_enhance_30_1-removebg-preview_j6mmhm.png" elif values.get("totalpoints") >=401 and values.get("totalpoints") <=600: values['badgeUrl'] ="https://res.cloudinary.com/dfmzougki/image/upload/v1742757797/last_level_LE_upscale_magic_x4_strength_20_similarity_50_tone_enhance_30_color_enhance_30_1-removebg-preview_bwmdyt.png" elif values.get("totalpoints") >=601 and values.get("totalpoints") <=800: values['badgeUrl'] ="https://res.cloudinary.com/dfmzougki/image/upload/v1742757797/image_25_LE_upscale_magic_x4_strength_20_similarity_50_1-removebg-preview_dcamov.png" elif values.get("totalpoints") >=801 and values.get("totalpoints") <=1000: values['badgeUrl'] ="https://res.cloudinary.com/dfmzougki/image/upload/v1742757797/Frame_118_LE_upscale_magic_x4_strength_20_similarity_50_tone_enhance_30_color_enhance_30_1-removebg-preview_ng6kzp.png" else: values['badgeUrl'] ="https://res.cloudinary.com/dfmzougki/image/upload/v1742757798/Frame_117_LE_upscale_magic_x4_strength_20_similarity_50_tone_enhance_30_color_enhance_30_1-removebg-preview_esrny1.png" return values