import json import sys import requests from typing import Dict, List, Any, Optional from collections import defaultdict, Counter import logging import math # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Ollama configuration OLLAMA_API_URL = "http://localhost:11434/api/generate" MODEL_NAME = "gemma3:12b" class ManagerCareerGuidanceProcessor: def __init__(self): self.skill_levels = { "1": "Beginner", "2": "Intermediate", "3": "Advanced" } self.total_tokens = 0 # Track total tokens def analyze_team_skills(self, users_data: Dict) -> Dict[str, Any]: """Analyze skills across the entire team""" skill_distribution = defaultdict(lambda: {"beginner": 0, "intermediate": 0, "advanced": 0}) all_skills = [] for user_id, user_info in users_data.get("users", {}).items(): for skill in user_info.get("skills", []): skill_name = skill.get("skill_name", "") skill_level = self.skill_levels.get(skill.get("skill_type", "1"), "Beginner") all_skills.append(skill_name) skill_distribution[skill_name][skill_level.lower()] += 1 # Get top skills in team skill_counts = Counter(all_skills) top_skills = [skill for skill, count in skill_counts.most_common(5)] return { "skill_distribution": dict(skill_distribution), "top_skills": top_skills, "total_unique_skills": len(skill_distribution), "skill_counts": dict(skill_counts) } def analyze_course_completion(self, users_data: Dict) -> Dict[str, Any]: """Analyze course completion patterns across the team""" total_assigned = 0 total_completed = 0 completion_by_user = {} for user_id, user_info in users_data.get("users", {}).items(): assigned = len(user_info.get("assignedCourses", [])) completed = len(user_info.get("completedCourses", [])) total_assigned += assigned total_completed += completed completion_rate = (completed / assigned * 100) if assigned > 0 else 0 completion_by_user[user_id] = { "user_name": user_info.get("user_name", ""), "assigned": assigned, "completed": completed, "completion_rate": completion_rate } overall_completion_rate = (total_completed / total_assigned * 100) if total_assigned > 0 else 0 return { "overall_completion_rate": overall_completion_rate, "total_assigned": total_assigned, "total_completed": total_completed, "completion_by_user": completion_by_user } def identify_skill_gaps(self, manager_data: Dict, users_data: Dict, available_courses: List) -> List[str]: """Identify skill gaps across the team and manager""" # Get manager's required skills from job profile manager_info = list(manager_data.values())[0] required_skills = set() if manager_info.get("jobProfile", {}).get("job_profile_skills"): for skill in manager_info["jobProfile"]["job_profile_skills"]: required_skills.add(skill.get("skill_name", "")) # Get all available skills from courses available_skills_from_courses = set() for course in available_courses: for skill in course.get("skills", []): available_skills_from_courses.add(skill.get("skill_name", "")) # Get current team skills current_team_skills = set() for user_id, user_info in users_data.get("users", {}).items(): for skill in user_info.get("skills", []): current_team_skills.add(skill.get("skill_name", "")) # Add manager skills for skill in manager_info.get("skills", []): current_team_skills.add(skill.get("skill_name", "")) # Identify gaps skill_gaps = list((required_skills | available_skills_from_courses) - current_team_skills) return skill_gaps[:10] # Return top 10 gaps def generate_team_recommendations(self, users_data: Dict, available_courses: List) -> List[Dict]: """Generate course recommendations for each team member""" recommendations = [] for user_id, user_info in users_data.get("users", {}).items(): user_skills = set(skill.get("skill_name", "") for skill in user_info.get("skills", [])) completed_courses = set(user_info.get("completedCourses", [])) assigned_courses = set(user_info.get("assignedCourses", [])) # Find relevant courses not yet taken recommended_courses = [] for course in available_courses: course_id = course.get("courseId", "") if course_id not in completed_courses and course_id not in assigned_courses: course_skills = set(skill.get("skill_name", "") for skill in course.get("skills", [])) # Check if course teaches new skills or improves existing ones if course_skills and (not course_skills.issubset(user_skills) or len(course_skills.intersection(user_skills)) > 0): # Instead of just appending course_id, append an object with both id and name recommended_courses.append({ "courseId": course_id, "courseName": course.get("name", "") }) # Determine development priority completion_rate = len(user_info.get("completedCourses", [])) / max(len(user_info.get("assignedCourses", [])), 1) * 100 if completion_rate < 50: priority = "High - Focus on completing assigned courses" elif len(user_info.get("skills", [])) < 3: priority = "Medium - Skill diversification needed" else: priority = "Low - Advanced skill development" recommendations.append({ "user_id": user_id, "user_name": user_info.get("user_name", ""), "recommended_courses": recommended_courses[:5], # Top 5 recommendations with both ID and name "skill_gaps": list(set(skill.get("skill_name", "") for course in available_courses for skill in course.get("skills", [])) - user_skills)[:5], "development_priority": priority, "career_path_suggestion": self.suggest_career_path(user_info, available_courses) }) return recommendations def suggest_career_path(self, user_info: Dict, available_courses: List) -> str: """Suggest career path based on user's current skills and available learning""" user_skills = [skill.get("skill_name", "") for skill in user_info.get("skills", [])] # Simple career path logic based on skills if any("Python" in skill or "Machine Learning" in skill or "Deep Learning" in skill for skill in user_skills): return "Data Science/AI Specialist Track" elif any("Javascript" in skill or "Angular" in skill for skill in user_skills): return "Frontend Development Track" elif any("Langchain" in skill for skill in user_skills): return "AI Application Development Track" else: return "General Technology Track - Explore foundational skills first" def generate_manager_insights(self, manager_data: Dict, users_data: Dict, team_analytics: Dict) -> List[Dict]: """Generate insights specifically for the manager""" insights = [] # Team performance insight completion_rate = team_analytics.get("overall_completion_rate", 0) if completion_rate < 60: insights.append({ "insight_type": "Performance", "title": "Team Course Completion Below Target", "description": f"Your team's course completion rate is {completion_rate:.1f}%, which is below the recommended 70%.", "priority": "High", "actionable_steps": [ "Schedule one-on-one sessions with team members having low completion rates", "Identify and address barriers preventing course completion", "Set up weekly check-ins to track progress", "Consider adjusting course load or deadlines" ] }) # Skill diversity insight team_skills = team_analytics.get("skill_distribution", {}) if len(team_skills) < 5: insights.append({ "insight_type": "Skills", "title": "Limited Skill Diversity in Team", "description": f"Your team currently has expertise in only {len(team_skills)} different skill areas.", "priority": "Medium", "actionable_steps": [ "Encourage team members to explore complementary skills", "Assign cross-functional learning opportunities", "Consider hiring for skill gaps that can't be filled through training" ] }) # Leadership development insight insights.append({ "insight_type": "Leadership", "title": "Leadership Development Opportunity", "description": "As a manager, continuous leadership skill development will enhance team performance.", "priority": "Medium", "actionable_steps": [ "Enroll in management and leadership courses", "Practice regular feedback sessions with team members", "Attend leadership workshops or conferences", "Implement team building activities" ] }) return insights def call_ollama_api(self, prompt: str) -> str: """Make API call to Ollama and track token usage""" try: payload = { "model": MODEL_NAME, "prompt": prompt, "stream": False, "options": { "temperature": 0.7, "top_p": 0.9, "max_tokens": 2000 } } response = requests.post(OLLAMA_API_URL, json=payload, timeout=60) response.raise_for_status() result = response.json() # Approximate token count (1 token ≈ 1 word) prompt_tokens = len(prompt.split()) completion_tokens = len(result.get("response", "").split()) self.total_tokens += prompt_tokens + completion_tokens return result.get("response", "").strip() except requests.exceptions.RequestException as e: logger.error(f"Error calling Ollama API: {e}") return "I apologize, but I'm currently unable to process your request due to a technical issue. Please try again later." def create_manager_prompt(self, request_data: Dict) -> str: """Create a comprehensive prompt for the manager's query""" manager_query = request_data.get("manager_query", "") manager_data = request_data.get("manager_data", {}) users_data = request_data.get("users_under_manager_data", {}) # Extract manager info manager_info = list(manager_data.values())[0] if manager_data else {} manager_name = manager_info.get("user_name", "Manager") manager_designation = manager_info.get("designation", "") # Team summary team_size = len(users_data.get("users", {})) prompt = f""" You are an expert career guidance counselor and management consultant. You're helping {manager_name}, who holds the position of {manager_designation} and manages a team of {team_size} people. Manager's Query: "{manager_query}" Context: - Manager: {manager_name} ({manager_designation}) - Team Size: {team_size} members - You have access to detailed information about the manager's skills, courses, and team members' progress As a professional career advisor, please provide: 1. A direct, actionable answer to the manager's specific query 2. Strategic insights for team management and development 3. Personal development recommendations for the manager 4. Team optimization suggestions Your response should be: - Professional and management-focused - Actionable with concrete next steps - Strategic in nature, considering both individual and team development - Considerate of the manager's leadership responsibilities Please provide a comprehensive response that addresses the query while offering valuable management insights. """ return prompt.strip() def generate_skill_distribution_chart(self, team_skills_analysis: Dict) -> Dict[str, Any]: """Generate skill distribution bar chart""" skill_counts = team_skills_analysis.get("skill_counts", {}) return { "chart_type": "bar", "title": "Team Skill Distribution", "data": [{"skill": skill, "count": count} for skill, count in skill_counts.items()], "labels": list(skill_counts.keys()), "colors": ["#3B82F6", "#10B981", "#F59E0B", "#EF4444", "#8B5CF6"], "description": "Distribution of skills across team members", "insights": [ f"Most common skill: {max(skill_counts.items(), key=lambda x: x[1])[0] if skill_counts else 'None'}", f"Total unique skills: {len(skill_counts)}", f"Skills needing attention: {[skill for skill, count in skill_counts.items() if count < 2]}" ], "chart_config": { "x_axis": "Skills", "y_axis": "Number of Team Members", "show_values": True } } def generate_completion_rate_chart(self, course_completion_analysis: Dict) -> Dict[str, Any]: """Generate course completion rate chart""" completion_by_user = course_completion_analysis.get("completion_by_user", {}) chart_data = [] for user_id, data in completion_by_user.items(): chart_data.append({ "user": data.get("user_name", f"User {user_id}"), "completion_rate": round(data.get("completion_rate", 0), 1), "completed": data.get("completed", 0), "assigned": data.get("assigned", 0) }) return { "chart_type": "bar", "title": "Individual Course Completion Rates", "data": chart_data, "labels": [item["user"] for item in chart_data], "colors": ["#10B981" if item["completion_rate"] >= 70 else "#F59E0B" if item["completion_rate"] >= 50 else "#EF4444" for item in chart_data], "description": "Course completion percentage by team member", "insights": [ f"Average completion rate: {course_completion_analysis.get('overall_completion_rate', 0):.1f}%", f"Top performer: {max(chart_data, key=lambda x: x['completion_rate'])['user'] if chart_data else 'None'}", f"Needs attention: {[item['user'] for item in chart_data if item['completion_rate'] < 50]}" ], "chart_config": { "x_axis": "Team Members", "y_axis": "Completion Rate (%)", "target_line": 70 } } def generate_skill_level_heatmap(self, team_skills_analysis: Dict) -> Dict[str, Any]: """Generate skill level heatmap""" skill_distribution = team_skills_analysis.get("skill_distribution", {}) heatmap_data = [] for skill, levels in skill_distribution.items(): heatmap_data.append({ "skill": skill, "beginner": levels.get("beginner", 0), "intermediate": levels.get("intermediate", 0), "advanced": levels.get("advanced", 0) }) return { "chart_type": "heatmap", "title": "Team Skill Proficiency Matrix", "data": heatmap_data, "labels": list(skill_distribution.keys()), "colors": ["#FEF3C7", "#FCD34D", "#F59E0B", "#D97706"], "description": "Skill proficiency levels across the team", "insights": [ f"Skills with advanced expertise: {[skill for skill, levels in skill_distribution.items() if levels.get('advanced', 0) > 0]}", f"Skills needing development: {[skill for skill, levels in skill_distribution.items() if levels.get('beginner', 0) > levels.get('advanced', 0)]}", "Focus on advancing intermediate skills to expert level" ], "chart_config": { "x_axis": "Skill Levels", "y_axis": "Skills", "color_scale": "YlOrRd" } } def generate_career_path_pie_chart(self, team_recommendations: List) -> Dict[str, Any]: """Generate career path distribution pie chart""" career_paths = {} for rec in team_recommendations: path = rec.get("career_path_suggestion", "Unknown") career_paths[path] = career_paths.get(path, 0) + 1 return { "chart_type": "pie", "title": "Team Career Path Distribution", "data": [{"path": path, "count": count, "percentage": round(count/len(team_recommendations)*100, 1)} for path, count in career_paths.items()], "labels": list(career_paths.keys()), "colors": ["#3B82F6", "#10B981", "#F59E0B", "#EF4444", "#8B5CF6", "#EC4899"], "description": "Distribution of suggested career paths for team members", "insights": [ f"Most common career path: {max(career_paths.items(), key=lambda x: x[1])[0] if career_paths else 'None'}", f"Career diversity score: {len(career_paths)}/5", "Consider cross-training opportunities between different tracks" ], "chart_config": { "show_percentages": True, "show_legends": True } } def generate_priority_matrix_scatter(self, team_recommendations: List) -> Dict[str, Any]: """Generate development priority scatter plot""" priority_map = {"High": 3, "Medium": 2, "Low": 1} scatter_data = [] for rec in team_recommendations: skill_gap_count = len(rec.get("skill_gaps", [])) priority_score = priority_map.get(rec.get("development_priority", "").split(" - ")[0], 1) scatter_data.append({ "user": rec.get("user_name", "Unknown"), "skill_gaps": skill_gap_count, "priority": priority_score, "priority_label": rec.get("development_priority", "Unknown") }) return { "chart_type": "scatter", "title": "Development Priority vs Skill Gaps", "data": scatter_data, "labels": [item["user"] for item in scatter_data], "colors": ["#EF4444" if item["priority"] == 3 else "#F59E0B" if item["priority"] == 2 else "#10B981" for item in scatter_data], "description": "Correlation between development priority and skill gaps", "insights": [ f"High priority team members: {len([item for item in scatter_data if item['priority'] == 3])}", f"Average skill gaps: {sum(item['skill_gaps'] for item in scatter_data) / len(scatter_data):.1f}", "Focus on high-priority, high-skill-gap individuals first" ], "chart_config": { "x_axis": "Number of Skill Gaps", "y_axis": "Development Priority", "bubble_size": "priority" } } def generate_team_performance_table(self, course_completion_analysis: Dict, team_recommendations: List) -> Dict[str, Any]: """Generate comprehensive team performance table""" completion_by_user = course_completion_analysis.get("completion_by_user", {}) headers = ["Team Member", "Courses Assigned", "Courses Completed", "Completion Rate", "Priority", "Skill Gaps", "Recommended Path"] rows = [] for rec in team_recommendations: user_id = rec.get("user_id", "") user_data = completion_by_user.get(user_id, {}) rows.append([ rec.get("user_name", "Unknown"), str(user_data.get("assigned", 0)), str(user_data.get("completed", 0)), f"{user_data.get('completion_rate', 0):.1f}%", rec.get("development_priority", "Unknown").split(" - ")[0], str(len(rec.get("skill_gaps", []))), rec.get("career_path_suggestion", "Unknown")[:30] + "..." if len(rec.get("career_path_suggestion", "")) > 30 else rec.get("career_path_suggestion", "Unknown") ]) return { "title": "Team Performance Overview", "headers": headers, "rows": rows, "description": "Comprehensive overview of team member performance and development needs", "insights": [ f"Team size: {len(rows)} members", f"Average completion rate: {course_completion_analysis.get('overall_completion_rate', 0):.1f}%", f"High priority members: {len([row for row in rows if 'High' in row[4]])}", "Regular performance reviews recommended for low completion rates" ], "styling": { "highlight_rows": [i for i, row in enumerate(rows) if float(row[3].replace('%', '')) < 50], "color_coding": { "completion_rate": {"good": 70, "warning": 50, "danger": 30} } } } def generate_skill_gap_analysis_table(self, skill_gaps: List, available_courses: List) -> Dict[str, Any]: """Generate skill gap analysis table""" headers = ["Skill Gap", "Severity", "Available Courses", "Recommended Action", "Timeline"] rows = [] for gap in skill_gaps[:10]: # Top 10 gaps matching_courses = [course.get("name", "") for course in available_courses if any(skill.get("skill_name", "") == gap for skill in course.get("skills", []))] severity = "High" if len(matching_courses) == 0 else "Medium" if len(matching_courses) < 2 else "Low" action = "External training needed" if len(matching_courses) == 0 else "Internal course available" timeline = "Immediate" if severity == "High" else "Next Quarter" rows.append([ gap, severity, ", ".join(matching_courses[:2]) + ("..." if len(matching_courses) > 2 else ""), action, timeline ]) return { "title": "Critical Skill Gap Analysis", "headers": headers, "rows": rows, "description": "Analysis of critical skill gaps and remediation strategies", "insights": [ f"Total skill gaps identified: {len(skill_gaps)}", f"High severity gaps: {len([row for row in rows if row[1] == 'High'])}", f"Gaps with available internal training: {len([row for row in rows if 'Internal' in row[3]])}", "Prioritize high-severity gaps for immediate action" ], "styling": { "highlight_rows": [i for i, row in enumerate(rows) if row[1] == "High"], "color_coding": { "severity": {"High": "#EF4444", "Medium": "#F59E0B", "Low": "#10B981"} } } } def generate_manager_dashboard_metrics(self, team_analytics: Dict, manager_insights: List) -> Dict[str, Any]: """Generate key metrics for manager dashboard""" return { "total_team_members": team_analytics.get("total_team_members", 0), "overall_completion_rate": round(team_analytics.get("course_completion_rate", 0), 1), "unique_skills": len(team_analytics.get("skill_distribution", {})), "skill_gaps": len(team_analytics.get("skill_gaps_identified", [])), "high_priority_alerts": len([insight for insight in manager_insights if insight.get("priority") == "High"]), "team_performance_score": self.calculate_team_performance_score(team_analytics), "recommended_actions": len([insight for insight in manager_insights]), "skill_diversity_index": self.calculate_skill_diversity_index(team_analytics.get("skill_distribution", {})) } def calculate_team_performance_score(self, team_analytics: Dict) -> int: """Calculate overall team performance score (0-100)""" completion_rate = team_analytics.get("course_completion_rate", 0) skill_diversity = len(team_analytics.get("skill_distribution", {})) # Weight completion rate (70%) and skill diversity (30%) score = (completion_rate * 0.7) + (min(skill_diversity * 10, 30) * 1.0) return min(int(score), 100) def calculate_skill_diversity_index(self, skill_distribution: Dict) -> float: """Calculate skill diversity index (0.0-1.0)""" if not skill_distribution: return 0.0 total_skills = sum(skill_distribution.values()) if total_skills == 0: return 0.0 # Calculate entropy-based diversity index diversity = 0.0 for count in skill_distribution.values(): if count > 0: p = count / total_skills if p > 0: diversity -= p * math.log2(p) if len(skill_distribution) > 1: max_diversity = math.log2(len(skill_distribution)) else: max_diversity = 1.0 return min(diversity / max_diversity if max_diversity > 0 else 0, 1.0) def generate_data_visualizations(self, team_skills_analysis: Dict, course_completion_analysis: Dict, team_recommendations: List, skill_gaps: List, available_courses: List, team_analytics: Dict, manager_insights: List) -> Dict[str, Any]: """Generate comprehensive data visualizations""" charts = [ self.generate_skill_distribution_chart(team_skills_analysis), self.generate_completion_rate_chart(course_completion_analysis), self.generate_skill_level_heatmap(team_skills_analysis), self.generate_career_path_pie_chart(team_recommendations), self.generate_priority_matrix_scatter(team_recommendations) ] tables = [ self.generate_team_performance_table(course_completion_analysis, team_recommendations), self.generate_skill_gap_analysis_table(skill_gaps, available_courses) ] summary_metrics = self.generate_manager_dashboard_metrics(team_analytics, manager_insights) visualization_insights = [ f"Team performance score: {summary_metrics['team_performance_score']}/100", f"Skill diversity index: {summary_metrics['skill_diversity_index']:.2f}", f"Critical areas need immediate attention: {summary_metrics['high_priority_alerts']} identified", f"Overall team readiness: {'Strong' if summary_metrics['team_performance_score'] > 75 else 'Developing' if summary_metrics['team_performance_score'] > 50 else 'Needs Improvement'}", "Data visualizations updated in real-time based on team progress" ] return { "charts": charts, "tables": tables, "summary_metrics": summary_metrics, "visualization_insights": visualization_insights } # METHODS START FOR MANAGER STRATEGIC ROADMAP def generate_manager_career_roadmap(self, manager_data: Dict, team_analytics: Dict, manager_insights: List) -> Dict[str, Any]: """Generate dynamic career roadmap for the manager based on actual data analysis""" manager_info = list(manager_data.values())[0] if manager_data else {} manager_skills = [skill.get("skill_name", "") for skill in manager_info.get("skills", [])] manager_designation = manager_info.get("designation", "") # Analyze manager's current state completion_rate = len(manager_info.get("completedCourses", [])) / max(len(manager_info.get("assignedCourses", [])), 1) * 100 team_performance = team_analytics.get("course_completion_rate", 0) high_priority_issues = len([insight for insight in manager_insights if insight.get("priority") == "High"]) # Generate manager growth path based on current skills and team needs manager_growth_areas = [] if "Javascript" in manager_skills or "Angular" in manager_skills: manager_growth_areas.append("Technical Leadership in Frontend Technologies") if "Python" in manager_skills or "Machine Learning" in manager_skills: manager_growth_areas.append("AI/ML Team Leadership and Strategy") if "Langchain" in manager_skills: manager_growth_areas.append("AI Application Development Leadership") # Add leadership areas based on team performance if team_performance < 60: manager_growth_areas.append("Performance Management and Team Motivation") if high_priority_issues > 2: manager_growth_areas.append("Strategic Problem Solving and Crisis Management") # Default leadership areas if len(manager_growth_areas) < 2: manager_growth_areas.extend([ "Strategic Team Development", "Advanced Leadership Communication" ]) manager_growth = " | ".join(manager_growth_areas[:3]) return { "manager_growth": manager_growth, "focus_areas": manager_growth_areas, "development_timeline": self.create_manager_development_timeline(manager_info, team_analytics), "leadership_priorities": self.identify_leadership_priorities(team_analytics, manager_insights) } def generate_team_development_roadmap(self, team_recommendations: List, team_analytics: Dict, skill_gaps: List) -> Dict[str, Any]: """Generate dynamic team development roadmap based on team analysis""" team_size = team_analytics.get("total_team_members", 0) completion_rate = team_analytics.get("course_completion_rate", 0) top_skills = team_analytics.get("top_skills_in_team", []) # Analyze team development needs development_priorities = [] # Priority 1: Course completion if low if completion_rate < 70: development_priorities.append(f"Improve course completion rate from {completion_rate:.1f}% to 80%+") # Priority 2: Skill diversification if len(top_skills) < 5: development_priorities.append("Expand skill diversity across team members") # Priority 3: Address critical skill gaps if len(skill_gaps) > 5: critical_gaps = skill_gaps[:3] development_priorities.append(f"Address critical skill gaps: {', '.join(critical_gaps)}") # Priority 4: Career path alignment career_paths = {} for rec in team_recommendations: path = rec.get("career_path_suggestion", "General") career_paths[path] = career_paths.get(path, 0) + 1 if len(career_paths) > 1: dominant_path = max(career_paths.items(), key=lambda x: x[1])[0] development_priorities.append(f"Align team development with {dominant_path} focus") # Create timeline-based roadmap quarterly_plan = self.create_quarterly_development_plan(team_recommendations, skill_gaps) return { "development_priorities": development_priorities, "quarterly_plan": quarterly_plan, "skill_development_focus": self.identify_skill_development_focus(team_recommendations, top_skills), "team_growth_metrics": self.define_team_growth_metrics(team_analytics) } def generate_organizational_impact_roadmap(self, manager_data: Dict, team_analytics: Dict, available_courses: List) -> Dict[str, Any]: """Generate organizational impact roadmap based on job profile and business needs""" manager_info = list(manager_data.values())[0] if manager_data else {} job_profile = manager_info.get("jobProfile", {}) required_skills = job_profile.get("job_profile_skills", []) if job_profile else [] # Analyze organizational alignment organizational_priorities = [] # Check alignment with job profile requirements if required_skills: required_skill_names = [skill.get("skill_name", "") for skill in required_skills] team_skills = team_analytics.get("skill_distribution", {}).keys() alignment_gap = set(required_skill_names) - set(team_skills) if alignment_gap: organizational_priorities.append(f"Align team skills with organizational requirements: {', '.join(list(alignment_gap)[:3])}") # Business impact areas if team_analytics.get("course_completion_rate", 0) > 80: organizational_priorities.append("Leverage high-performing team for strategic initiatives") # Innovation and growth opportunities advanced_skills = ["Machine Learning", "Deep Learning", "AI", "Langchain"] team_advanced_skills = [skill for skill in team_analytics.get("skill_distribution", {}).keys() if any(adv in skill for adv in advanced_skills)] if team_advanced_skills: organizational_priorities.append(f"Drive innovation initiatives using team's {', '.join(team_advanced_skills[:2])} expertise") # Scale and efficiency improvements organizational_priorities.append("Implement knowledge sharing processes to scale team expertise") return { "strategic_alignment": organizational_priorities, "business_impact_areas": self.identify_business_impact_areas(team_analytics, available_courses), "innovation_opportunities": self.identify_innovation_opportunities(team_analytics), "scalability_plan": self.create_scalability_plan(team_analytics, manager_info) } def create_manager_development_timeline(self, manager_info: Dict, team_analytics: Dict) -> Dict[str, List[str]]: """Create a timeline-based development plan for the manager""" return { "immediate_0_30_days": [ "Conduct one-on-one sessions with underperforming team members", "Review and optimize current course assignments", "Identify immediate skill gaps and training needs" ], "short_term_1_3_months": [ "Implement performance tracking and regular feedback cycles", "Enroll in advanced leadership development programs", "Establish team skill development roadmaps" ], "medium_term_3_6_months": [ "Lead strategic initiatives leveraging team's core strengths", "Develop cross-functional collaboration frameworks", "Create knowledge sharing and mentorship programs" ], "long_term_6_12_months": [ "Drive organizational innovation projects", "Establish center of excellence for team's specialized skills", "Mentor other managers and scale successful practices" ] } def identify_leadership_priorities(self, team_analytics: Dict, manager_insights: List) -> List[str]: """Identify specific leadership priorities based on team analysis""" priorities = [] completion_rate = team_analytics.get("course_completion_rate", 0) if completion_rate < 60: priorities.append("Team Performance Management") high_priority_issues = [insight for insight in manager_insights if insight.get("priority") == "High"] if len(high_priority_issues) > 1: priorities.append("Crisis Management and Problem Resolution") team_size = team_analytics.get("total_team_members", 0) if team_size > 5: priorities.append("Large Team Leadership and Delegation") priorities.extend([ "Strategic Planning and Vision Setting", "Talent Development and Career Coaching", "Cross-functional Collaboration Leadership" ]) return priorities[:4] # Return top 4 priorities def create_quarterly_development_plan(self, team_recommendations: List, skill_gaps: List) -> Dict[str, Dict[str, Any]]: """Create a quarterly development plan for the team""" high_priority_members = [rec for rec in team_recommendations if rec.get("development_priority", "").startswith("High")] return { "q1_focus": { "primary_goal": "Foundation Building and Performance Improvement", "target_members": len(high_priority_members), "key_activities": [ "Complete assigned courses with <50% completion rate", "Address top 3 skill gaps identified", "Establish regular learning check-ins" ], "success_metrics": ["70%+ course completion rate", "Reduce skill gaps by 50%"] }, "q2_focus": { "primary_goal": "Skill Diversification and Specialization", "target_members": len(team_recommendations) - len(high_priority_members), "key_activities": [ "Begin specialized career path training", "Cross-training in complementary skills", "Start advanced certification programs" ], "success_metrics": ["2+ new skills per team member", "Career path clarity for all members"] }, "q3_focus": { "primary_goal": "Advanced Development and Leadership Preparation", "key_activities": [ "Lead internal knowledge sharing sessions", "Participate in cross-functional projects", "Begin mentoring junior team members" ], "success_metrics": ["Internal presentations completed", "Peer mentoring established"] }, "q4_focus": { "primary_goal": "Innovation and Strategic Contribution", "key_activities": [ "Lead innovation projects using new skills", "Contribute to organizational best practices", "Prepare for next-level responsibilities" ], "success_metrics": ["Innovation project completion", "Strategic impact demonstration"] } } def identify_skill_development_focus(self, team_recommendations: List, top_skills: List) -> Dict[str, Any]: """Identify specific skill development focus areas""" skill_categories = { "technical_skills": [], "emerging_technologies": [], "foundational_skills": [] } for rec in team_recommendations: skill_gaps = rec.get("skill_gaps", []) for gap in skill_gaps: if any(tech in gap.lower() for tech in ["ai", "machine learning", "deep learning", "langchain"]): skill_categories["emerging_technologies"].append(gap) elif any(tech in gap.lower() for tech in ["python", "javascript", "angular"]): skill_categories["technical_skills"].append(gap) else: skill_categories["foundational_skills"].append(gap) return { "priority_focus": "emerging_technologies" if skill_categories["emerging_technologies"] else "technical_skills", "development_sequence": [ "foundational_skills", "technical_skills", "emerging_technologies" ], "skill_categories": skill_categories } def define_team_growth_metrics(self, team_analytics: Dict) -> Dict[str, Any]: """Define measurable team growth metrics""" current_completion = team_analytics.get("course_completion_rate", 0) current_skills = len(team_analytics.get("skill_distribution", {})) return { "completion_rate_targets": { "current": f"{current_completion:.1f}%", "target_q1": f"{min(current_completion + 15, 85):.1f}%", "target_q2": "90%+", "stretch_goal": "95%+" }, "skill_diversity_targets": { "current": current_skills, "target_q1": current_skills + 2, "target_q2": current_skills + 4, "stretch_goal": current_skills + 6 }, "performance_indicators": [ "Course completion velocity", "Skill acquisition rate", "Team collaboration score", "Innovation project contributions" ] } def identify_business_impact_areas(self, team_analytics: Dict, available_courses: List) -> List[str]: """Identify areas where the team can create business impact""" impact_areas = [] # Based on current team skills team_skills = list(team_analytics.get("skill_distribution", {}).keys()) if any("machine learning" in skill.lower() or "ai" in skill.lower() for skill in team_skills): impact_areas.append("AI-driven process automation and optimization") if any("javascript" in skill.lower() or "angular" in skill.lower() for skill in team_skills): impact_areas.append("User experience enhancement and frontend innovation") if any("python" in skill.lower() for skill in team_skills): impact_areas.append("Data analysis and business intelligence solutions") # Default business impact areas impact_areas.extend([ "Cross-functional project leadership", "Knowledge transfer and best practices development", "Process improvement and efficiency optimization" ]) return impact_areas[:4] def identify_innovation_opportunities(self, team_analytics: Dict) -> List[str]: """Identify innovation opportunities based on team capabilities""" opportunities = [] team_skills = list(team_analytics.get("skill_distribution", {}).keys()) advanced_skills = ["Machine Learning", "Deep Learning", "Langchain", "AI"] team_advanced = [skill for skill in team_skills if any(adv in skill for adv in advanced_skills)] if team_advanced: opportunities.extend([ f"Develop internal AI tools using {', '.join(team_advanced[:2])} expertise", "Create proof-of-concept solutions for business challenges", "Lead organization's digital transformation initiatives" ]) opportunities.extend([ "Establish internal training and certification programs", "Develop best practices frameworks for other teams", "Create cross-departmental collaboration models" ]) return opportunities[:4] def create_scalability_plan(self, team_analytics: Dict, manager_info: Dict) -> Dict[str, Any]: """Create a plan for scaling team expertise and impact""" team_size = team_analytics.get("total_team_members", 0) return { "knowledge_scaling": { "documentation": "Create comprehensive skill and process documentation", "training_programs": "Develop internal training curricula", "mentorship": "Establish junior-senior mentorship programs" }, "team_expansion": { "hiring_priorities": f"Target skills to complement existing {list(team_analytics.get('top_skills_in_team', []))[:2]}", "onboarding": "Streamlined skill-based onboarding process", "succession_planning": "Identify and develop next-level leaders" }, "impact_multiplication": { "best_practices": "Document and share successful team practices", "consulting": "Provide expertise to other organizational units", "thought_leadership": "Represent organization in industry forums" } } def calculate_course_match_score(self, course: Dict, team_analytics: Dict, skill_gaps: List[str], manager_info: Dict) -> Dict[str, Any]: """Calculate comprehensive match score for a course with detailed reasoning""" score = 0 reasons = [] detailed_breakdown = {} course_skills = [skill.get("skill_name", "") for skill in course.get("skills", [])] team_skills = list(team_analytics.get("skill_distribution", {}).keys()) manager_skills = [skill.get("skill_name", "") for skill in manager_info.get("skills", [])] # 1. Skill Gap Coverage (35% weight) skill_gap_coverage = len(set(course_skills) & set(skill_gaps)) if skill_gap_coverage > 0: gap_score = min(skill_gap_coverage * 15, 35) score += gap_score detailed_breakdown["skill_gap_coverage"] = gap_score reasons.append(f"Addresses {skill_gap_coverage} critical skill gap(s): {', '.join(set(course_skills) & set(skill_gaps))}") # 2. Team Skill Enhancement (25% weight) team_skill_enhancement = len(set(course_skills) & set(team_skills)) if team_skill_enhancement > 0: enhancement_score = min(team_skill_enhancement * 8, 25) score += enhancement_score detailed_breakdown["team_skill_enhancement"] = enhancement_score reasons.append(f"Enhances existing team expertise in {', '.join(set(course_skills) & set(team_skills))}") # 3. New Skill Introduction (20% weight) new_skills = set(course_skills) - set(team_skills) - set(manager_skills) if new_skills: new_skill_score = min(len(new_skills) * 10, 20) score += new_skill_score detailed_breakdown["new_skill_introduction"] = new_skill_score reasons.append(f"Introduces valuable new skills: {', '.join(list(new_skills)[:2])}") # 4. Strategic Alignment (15% weight) strategic_skills = ["Machine Learning", "Deep Learning", "AI", "Python", "Langchain", "Angular", "Javascript"] strategic_match = len(set(course_skills) & set(strategic_skills)) if strategic_match > 0: strategic_score = min(strategic_match * 7, 15) score += strategic_score detailed_breakdown["strategic_alignment"] = strategic_score reasons.append(f"Aligns with strategic technology focus: {', '.join(set(course_skills) & set(strategic_skills))}") # 5. Team Completion Readiness (5% weight) team_completion_rate = team_analytics.get("course_completion_rate", 0) if team_completion_rate > 70: readiness_score = 5 score += readiness_score detailed_breakdown["completion_readiness"] = readiness_score reasons.append("Team shows strong course completion capability") elif team_completion_rate > 50: readiness_score = 3 score += readiness_score detailed_breakdown["completion_readiness"] = readiness_score reasons.append("Team has moderate completion readiness") # Bonus points for comprehensive courses if len(course_skills) >= 3: score += 2 detailed_breakdown["comprehensive_bonus"] = 2 reasons.append("Comprehensive course covering multiple related skills") # Ensure minimum reasoning if not reasons: reasons.append("Foundational course for general skill development") score = max(score, 10) # Minimum score for any course return { "match_score": min(score, 100), # Cap at 100 "reasons": reasons, "detailed_breakdown": detailed_breakdown, "recommendation_strength": "High" if score >= 70 else "Medium" if score >= 40 else "Low" } def generate_top_course_recommendations(self, manager_data: Dict, users_data: Dict, available_courses: List, team_analytics: Dict, skill_gaps: List) -> List[Dict]: """Generate exactly 9 top course recommendations with match scores and detailed reasons""" manager_info = list(manager_data.values())[0] if manager_data else {} manager_completed = set(manager_info.get("completedCourses", [])) manager_assigned = set(manager_info.get("assignedCourses", [])) # Get all team member completed and assigned courses to avoid duplication team_completed = set() team_assigned = set() for user_info in users_data.get("users", {}).values(): team_completed.update(user_info.get("completedCourses", [])) team_assigned.update(user_info.get("assignedCourses", [])) course_recommendations = [] for course in available_courses: course_id = course.get("courseId", "") # Skip if already completed or assigned to manager or team if course_id in manager_completed or course_id in manager_assigned: continue # Calculate match score and reasoning match_analysis = self.calculate_course_match_score( course, team_analytics, skill_gaps, manager_info ) # Determine specific benefit for manager manager_benefit = self.determine_manager_specific_benefit(course, manager_info, team_analytics) # Create comprehensive recommendation recommendation = { "courseId": course_id, "courseName": course.get("name", ""), "shortDescription": course.get("short_description", ""), "matchScore": match_analysis["match_score"], "recommendationStrength": match_analysis["recommendation_strength"], "primaryReasons": match_analysis["reasons"][:3], # Top 3 reasons "detailedReasons": { "skill_development": self.generate_skill_development_reason(course, skill_gaps), "team_impact": self.generate_team_impact_reason(course, team_analytics), "strategic_value": self.generate_strategic_value_reason(course, manager_info), "implementation_priority": self.generate_implementation_priority(course, team_analytics) }, "managerSpecificBenefit": manager_benefit, "skillsCovered": [skill.get("skill_name", "") for skill in course.get("skills", [])], "scoringBreakdown": match_analysis["detailed_breakdown"], "recommendedFor": "Team Development and Strategic Growth", "estimatedImpact": self.estimate_course_impact(course, team_analytics), "implementationTimeline": self.suggest_implementation_timeline(match_analysis["match_score"]) } course_recommendations.append(recommendation) # Sort by match score and take top 9 course_recommendations.sort(key=lambda x: x["matchScore"], reverse=True) # Ensure we have exactly 9 recommendations top_9_recommendations = course_recommendations[:9] # If we have fewer than 9, pad with strategic reasons for available courses while len(top_9_recommendations) < 9 and len(top_9_recommendations) < len(available_courses): remaining_courses = available_courses[len(top_9_recommendations):] if remaining_courses: course = remaining_courses[0] course_id = course.get("courseId", "") if course_id not in [rec["courseId"] for rec in top_9_recommendations]: padded_recommendation = { "courseId": course_id, "courseName": course.get("name", ""), "shortDescription": course.get("short_description", ""), "matchScore": 25, # Base score for padding "recommendationStrength": "Low", "primaryReasons": ["Foundational skill development", "Future team capability building"], "detailedReasons": { "skill_development": "Provides foundational knowledge for team growth", "team_impact": "Builds baseline competency for future projects", "strategic_value": "Supports long-term team development strategy", "implementation_priority": "Consider for future quarters" }, "managerSpecificBenefit": "Expands team's overall knowledge base", "skillsCovered": [skill.get("skill_name", "") for skill in course.get("skills", [])], "scoringBreakdown": {"foundational_value": 25}, "recommendedFor": "Long-term Team Development", "estimatedImpact": "Low to Medium", "implementationTimeline": "Future Consideration" } top_9_recommendations.append(padded_recommendation) else: break return top_9_recommendations[:9] # Ensure exactly 9 def determine_manager_specific_benefit(self, course: Dict, manager_info: Dict, team_analytics: Dict) -> str: """Determine specific benefit for the manager in recommending this course""" course_skills = [skill.get("skill_name", "") for skill in course.get("skills", [])] team_completion_rate = team_analytics.get("course_completion_rate", 0) if team_completion_rate < 60: return f"Assigning this course can help improve team's {team_completion_rate:.1f}% completion rate while building critical {', '.join(course_skills[:2])} capabilities" elif any("Machine Learning" in skill or "AI" in skill for skill in course_skills): return "Positions your team at the forefront of AI technology, enhancing department's innovation capacity and competitive advantage" elif any("Javascript" in skill or "Angular" in skill for skill in course_skills): return "Strengthens team's frontend development capabilities, enabling better user experience delivery and client satisfaction" elif any("Python" in skill for skill in course_skills): return "Builds versatile programming foundation, enabling team to handle diverse technical challenges and data analysis tasks" else: return f"Diversifies team skill portfolio with {', '.join(course_skills)}, reducing technical dependencies and increasing project flexibility" def generate_skill_development_reason(self, course: Dict, skill_gaps: List[str]) -> str: """Generate detailed skill development reasoning""" course_skills = [skill.get("skill_name", "") for skill in course.get("skills", [])] gap_coverage = set(course_skills) & set(skill_gaps) if gap_coverage: return f"Directly addresses {len(gap_coverage)} identified skill gaps ({', '.join(gap_coverage)}), reducing team's technical vulnerabilities and improving project delivery capabilities." else: return f"Enhances team's technical depth in {', '.join(course_skills[:2])}, providing advanced capabilities for complex project requirements." def generate_team_impact_reason(self, course: Dict, team_analytics: Dict) -> str: """Generate team impact reasoning""" team_size = team_analytics.get("total_team_members", 0) completion_rate = team_analytics.get("course_completion_rate", 0) if completion_rate > 80: return f"With your team's strong {completion_rate:.1f}% completion rate, this course will effectively upskill all {team_size} members, multiplying the knowledge impact across the team." elif completion_rate > 60: return f"Moderate completion rate ({completion_rate:.1f}%) suggests selective assignment to high-performing team members first, then cascading knowledge to others." else: return f"Current {completion_rate:.1f}% completion rate indicates need for engaging, practical courses like this to rebuild learning momentum and team confidence." def generate_strategic_value_reason(self, course: Dict, manager_info: Dict) -> str: """Generate strategic value reasoning""" course_skills = [skill.get("skill_name", "") for skill in course.get("skills", [])] manager_designation = manager_info.get("designation", "Manager") strategic_skills = ["Machine Learning", "Deep Learning", "AI", "Langchain"] if any(skill in strategic_skills for skill in course_skills): return f"As {manager_designation}, investing in AI/ML capabilities positions your team for high-impact projects, potential budget increases, and organizational recognition." else: return f"Strengthens your team's foundational capabilities, enabling you as {manager_designation} to take on more diverse projects and demonstrate team versatility to senior leadership." def generate_implementation_priority(self, course: Dict, team_analytics: Dict) -> str: """Generate implementation priority guidance""" completion_rate = team_analytics.get("course_completion_rate", 0) course_skills = [skill.get("skill_name", "") for skill in course.get("skills", [])] if completion_rate > 80: return "High priority - Team ready for immediate implementation with strong completion likelihood" elif completion_rate > 60: return "Medium priority - Assign to top performers first, then expand based on initial success" elif any("foundational" in skill.lower() or "basic" in skill.lower() for skill in course_skills): return "High priority - Foundational skills needed to improve overall team performance" else: return "Low priority - Focus on completing existing assignments first before adding advanced courses" def estimate_course_impact(self, course: Dict, team_analytics: Dict) -> str: """Estimate the potential impact of the course""" course_skills = [skill.get("skill_name", "") for skill in course.get("skills", [])] team_skills = list(team_analytics.get("skill_distribution", {}).keys()) skill_overlap = len(set(course_skills) & set(team_skills)) if skill_overlap >= 2: return "High Impact - Builds on existing team strengths" elif len(course_skills) >= 3: return "High Impact - Comprehensive skill development" elif skill_overlap == 1: return "Medium Impact - Enhances current capabilities" else: return "Medium Impact - Introduces new capabilities" def suggest_implementation_timeline(self, match_score: int) -> str: """Suggest implementation timeline based on match score""" if match_score >= 80: return "Immediate (This Quarter)" elif match_score >= 60: return "Short-term (Next Quarter)" elif match_score >= 40: return "Medium-term (Within 6 months)" else: return "Long-term (Future consideration)" def process_career_guidance(self, request_data: Dict) -> Dict[str, Any]: """Main processing function for manager career guidance""" try: # Reset token counter for each new request self.total_tokens = 0 # Extract data manager_data = request_data.get("manager_data", {}) users_data = request_data.get("users_under_manager_data", {}) available_courses = request_data.get("client_all_courses_data", {}).get(str(request_data.get("client_id", "")), []) # Analyze team team_skills_analysis = self.analyze_team_skills(users_data) course_completion_analysis = self.analyze_course_completion(users_data) skill_gaps = self.identify_skill_gaps(manager_data, users_data, available_courses) # Generate recommendations team_recommendations = self.generate_team_recommendations(users_data, available_courses) # Create team analytics team_analytics = { "total_team_members": len(users_data.get("users", {})), "skill_distribution": team_skills_analysis.get("skill_counts", {}), "course_completion_rate": course_completion_analysis.get("overall_completion_rate", 0), "top_skills_in_team": team_skills_analysis.get("top_skills", []), "skill_gaps_identified": skill_gaps, "team_performance_summary": f"Team of {len(users_data.get('users', {}))} members with {course_completion_analysis.get('overall_completion_rate', 0):.1f}% course completion rate" } # Generate manager insights manager_insights = self.generate_manager_insights(manager_data, users_data, team_analytics) # Get AI response for the specific query prompt = self.create_manager_prompt(request_data) ai_response = self.call_ollama_api(prompt) suggested_courses_for_manager = self.generate_top_course_recommendations( manager_data, users_data, available_courses, team_analytics, skill_gaps ) # Leadership development suggestions leadership_suggestions = [ "Develop coaching and mentoring skills to better support team members", "Learn about performance management and goal setting frameworks", "Study team dynamics and conflict resolution techniques", "Explore strategic thinking and decision-making methodologies", "Practice effective communication and feedback delivery" ] # Generate comprehensive data visualizations data_visualizations = self.generate_data_visualizations( team_skills_analysis, course_completion_analysis, team_recommendations, skill_gaps, available_courses, team_analytics, manager_insights ) manager_roadmap = self.generate_manager_career_roadmap(manager_data, team_analytics, manager_insights) team_roadmap = self.generate_team_development_roadmap(team_recommendations, team_analytics, skill_gaps) organizational_roadmap = self.generate_organizational_impact_roadmap(manager_data, team_analytics, available_courses) career_roadmap_suggestions = { "manager_growth": manager_roadmap, "team_development": team_roadmap, "organizational_impact": organizational_roadmap } return { "answer": ai_response, "manager_insights": manager_insights, "team_recommendations": team_recommendations, "team_analytics": team_analytics, "suggested_courses_for_manager": suggested_courses_for_manager, "leadership_development_suggestions": leadership_suggestions, "team_skill_matrix": team_skills_analysis, "career_roadmap_suggestions": career_roadmap_suggestions, "data_visualizations": data_visualizations, "total_tokens": self.total_tokens } except Exception: return { "answer": "I apologize, but there was an error processing your career guidance request. Please try again.", "manager_insights": [], "team_recommendations": [], "team_analytics": { "total_team_members": 0, "skill_distribution": {}, "course_completion_rate": 0.0, "top_skills_in_team": [], "skill_gaps_identified": [], "team_performance_summary": "" }, "suggested_courses_for_manager": [], "leadership_development_suggestions": [], "team_skill_matrix": {}, "career_roadmap_suggestions": {}, "data_visualizations": { "charts": [], "tables": [], "summary_metrics": {}, "visualization_insights": [] }, "total_tokens": 0 } def main(request_data=None): """ Main function to process manager career guidance requests. Args: request_data (dict, optional): Request data passed programmatically """ # Only use command line arguments if no request_data is provided if request_data is None: if len(sys.argv) != 2: print("Usage: python manager_career_guidance_processor.py ") sys.exit(1) json_file_path = sys.argv[1] try: with open(json_file_path, 'r') as file: request_data = json.load(file) except FileNotFoundError: error_response = create_error_response("File not found") print(json.dumps(error_response)) sys.exit(0) except json.JSONDecodeError: error_response = create_error_response("Invalid JSON format") print(json.dumps(error_response)) sys.exit(0) except Exception as e: error_response = create_error_response(f"Error reading file: {str(e)}") print(json.dumps(error_response)) sys.exit(0) # Process the request data try: processor = ManagerCareerGuidanceProcessor() result = processor.process_career_guidance(request_data) print(json.dumps(result)) except Exception as e: error_response = create_error_response(f"Processing error: {str(e)}") print(json.dumps(error_response)) def create_error_response(error_message="An error occurred while processing your request."): """ Create a standardized error response structure. Args: error_message (str): Custom error message Returns: dict: Error response structure """ return { "answer": error_message, "manager_insights": [], "team_recommendations": [], "team_analytics": { "total_team_members": 0, "skill_distribution": {}, "course_completion_rate": 0.0, "top_skills_in_team": [], "skill_gaps_identified": [], "team_performance_summary": "" }, "suggested_courses_for_manager": [], "leadership_development_suggestions": [], "team_skill_matrix": {}, "career_roadmap_suggestions": {}, "data_visualizations": { "charts": [], "tables": [], "summary_metrics": {}, "visualization_insights": [] } } # Test payload - only used when running the script directly for testing def get_test_payload(): """ Get test payload for development and testing purposes. This is only used when the script is run directly, not when called from FastAPI. """ return { "client_id": "3", "manager_query": "How can I improve my team's AI and machine learning capabilities? What should be our learning priorities for the next quarter?", "manager_data": { "244": { "user_name": "Kunal Jain", "managerId": "29", "designation": "Engineer", "jobProfile": { "job_profile_name": "data scientist", "job_profile_skills": [ { "skill_id": "20", "skill_type": "1", "skill_name": "Machine Learning" }, { "skill_id": "15", "skill_type": "1", "skill_name": "Python" }, { "skill_id": "17", "skill_type": "1", "skill_name": "Deep Learning" } ] }, "assignedCourses": [ "175", "174", "177", "178", "109" ], "completedCourses": [ "177" ], "skills": [ { "skill_id": "19", "skill_type": "1", "skill_name": "Angular " }, { "skill_id": "14", "skill_type": "1", "skill_name": "Javascript" }, { "skill_id": "16", "skill_type": "1", "skill_name": "Langchain" } ] } }, "users_under_manager_data": { "users": { "221": { "user_name": "Tanay Jain", "managerId": "244", "designation": "", "jobProfile": { "job_profile_name": "", "job_profile_skills": [] }, "assignedCourses": [ "106", "109", "110", "107", "111" ], "completedCourses": [ "109", "110", "111" ], "skills": [ { "skill_id": "17", "skill_type": "1", "skill_name": "Deep Learning" }, { "skill_id": "14", "skill_type": "2", "skill_name": "Javascript" }, { "skill_id": "15", "skill_type": "2", "skill_name": "Python" }, { "skill_id": "16", "skill_type": "3", "skill_name": "Langchain" } ] }, "223": { "user_name": "Kunal Jain", "managerId": "244", "designation": "", "jobProfile": { "job_profile_name": "", "job_profile_skills": [] }, "assignedCourses": [ "106", "109" ], "completedCourses": [ "109", "164", "178" ], "skills": [ { "skill_id": "14", "skill_type": "3", "skill_name": "Javascript" }, { "skill_id": "16", "skill_type": "3", "skill_name": "Langchain" } ] }, "243": { "user_name": "Kunal new", "managerId": "244", "designation": "", "jobProfile": { "job_profile_name": "", "job_profile_skills": [] }, "assignedCourses": [ "162", "163", "164", "167", "177", "178" ], "completedCourses": [], "skills": [ { "skill_id": "19", "skill_type": "1", "skill_name": "Angular " } ] }, "245": { "user_name": "Kunal Jain", "managerId": "244", "designation": "", "jobProfile": { "job_profile_name": "", "job_profile_skills": [] }, "assignedCourses": [ "177", "178" ], "completedCourses": [ "178" ], "skills": [ { "skill_id": "14", "skill_type": "1", "skill_name": "Javascript" } ] }, "246": { "user_name": "Kunal Jain", "managerId": "244", "designation": "", "jobProfile": { "job_profile_name": "", "job_profile_skills": [] }, "assignedCourses": [], "completedCourses": [], "skills": [] } } }, "client_all_courses_data": { "3": [ { "courseId": "43", "name": "Python Programming for Complete Beginners ", "short_description": "Python Programming for Complete Beginners ", "description": "", "skills": [] }, { "courseId": "60", "name": "New CCourse ", "short_description": "New COurse short desc", "description": "", "skills": [ { "skill_id": "14", "skill_type": "2", "skill_name": "Javascript" } ] }, { "courseId": "110", "name": "Langchain framework- Cycle - 1", "short_description": "Future of next AI genration", "description": "

LangChain is an open-source framework for developing applications that use large language models (LLMs).", "skills": [ { "skill_id": "16", "skill_type": "1", "skill_name": "Langchain" }, { "skill_id": "17", "skill_type": "1", "skill_name": "Deep Learning" }, { "skill_id": "16", "skill_type": "3", "skill_name": "Langchain" } ] }, { "courseId": "175", "name": "Angular intermmediate level course", "short_description": "Angular intermmediate level course", "description": "

Angular intermmediate level course

", "skills": [ { "skill_id": "19", "skill_type": "2", "skill_name": "Angular " } ] } ] } } if __name__ == "__main__": # When script is run directly (for testing), use test payload if no file is provided if len(sys.argv) == 1: print("Running with test payload for development testing...") test_payload = get_test_payload() main(test_payload) else: # Normal execution with command line arguments main()